Merge pull request 'Queue multiple offline Week-to-Today pulls' (#1250) from timmy/1249-queue-offline-week-pulls into main
This commit is contained in:
commit
a4498fdbbc
|
|
@ -391,7 +391,7 @@
|
|||
});
|
||||
const weekPlan = createWeekPlan({
|
||||
fetchJson:fetchReviewJson,localDate:todayRollover.localDate,timeZone:todayRollover.timeZone,
|
||||
storage:localStorage,getLogin:() => planningOwnerLogin,
|
||||
storage:localStorage,getLogin:() => planningOwnerLogin,coordinator:outboxCoordinator,
|
||||
});
|
||||
function openWeekPlanner(trigger) { planningTomorrow=false; return weekFlow.open(trigger); }
|
||||
function renderTomorrowQueueSummary(value) {
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>Date.now()}={}) {
|
||||
function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,coordinator,now=()=>Date.now()}={}) {
|
||||
let week={revision:0,timezone:null,days:[]};
|
||||
let flushing=null;
|
||||
let lastConflict=null;
|
||||
|
|
@ -39,20 +39,33 @@ function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>D
|
|||
const login=String(getLogin?.()||'').trim().toLowerCase();
|
||||
return login?pullPrefix+encodeURIComponent(login):'';
|
||||
}
|
||||
function pendingPull() {
|
||||
function pullQueue() {
|
||||
const key=pullKey();
|
||||
if(!key||!storage)return null;
|
||||
if(!key||!storage)return [];
|
||||
try {
|
||||
const value=JSON.parse(storage.getItem(key)||'null');
|
||||
return typeof value?.body?.operation_id==='string'&&Array.isArray(value?.today?.ids)&&
|
||||
Number.isInteger(value?.week?.revision)&&Array.isArray(value?.week?.days)?value:null;
|
||||
} catch(_error){return null;}
|
||||
const entries=Array.isArray(value?.entries)?value.entries:(value?[value]:[]);
|
||||
return entries.filter(entry=>typeof entry?.body?.operation_id==='string'&&Array.isArray(entry?.today?.ids)&&
|
||||
Number.isInteger(entry?.week?.revision)&&Array.isArray(entry?.week?.days));
|
||||
} catch(_error){return [];}
|
||||
}
|
||||
function writePullQueue(entries) {
|
||||
const key=pullKey();if(!key||!storage)return false;
|
||||
try {
|
||||
if(entries.length)storage.setItem(key,JSON.stringify({version:2,entries}));
|
||||
else storage.removeItem(key);
|
||||
return true;
|
||||
} catch(_error){return false;}
|
||||
}
|
||||
function pendingPull() {
|
||||
return pullQueue()[0]||null;
|
||||
}
|
||||
function resumePull() {
|
||||
const value=pendingPull();
|
||||
const queue=pullQueue(),conflicted=queue.find(entry=>entry.conflict&&entry.current?.week);
|
||||
const value=conflicted||queue[queue.length-1];
|
||||
if(!value)return null;
|
||||
adopt(value.week);
|
||||
return {...value,sync_pending:true};
|
||||
adopt(conflicted?conflicted.current.week:value.week);
|
||||
return {...value,...(conflicted?.current||{}),sync_pending:true};
|
||||
}
|
||||
function readConfirmed() {
|
||||
const key=confirmedKey();
|
||||
|
|
@ -502,8 +515,8 @@ function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>D
|
|||
return {...cloneDay(day),ids:(day.ids||[]).filter(id=>id!==identity),estimates};
|
||||
})};
|
||||
const record={body,today:optimisticToday,week:optimisticWeek,queued_at:now()};
|
||||
try {storage.setItem(key,JSON.stringify(record));}
|
||||
catch(_error){throw new Error('Could not save this pull on this device. Nothing changed.');}
|
||||
const queue=pullQueue();
|
||||
if(!writePullQueue([...queue,record]))throw new Error('Could not save this pull on this device. Nothing changed.');
|
||||
adopt(optimisticWeek);
|
||||
try{return await flushPull();}
|
||||
catch(error){
|
||||
|
|
@ -516,25 +529,44 @@ function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>D
|
|||
}
|
||||
function flushPull() {
|
||||
if(pulling)return pulling;
|
||||
const record=pendingPull(),key=pullKey();
|
||||
if(!record||!key)return Promise.resolve(false);
|
||||
pulling=(async()=>{
|
||||
let result;
|
||||
try {
|
||||
result=await fetchJson('api/v1/week/pull-item',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(record.body)});
|
||||
} catch(error) {
|
||||
if(error?.status!==409)throw error;
|
||||
const [today,currentWeek]=await Promise.all([fetchJson('api/v1/today'),fetchJson('api/v1/week')]);
|
||||
const conflicted={...record,conflict:true,current:{today,week:currentWeek}};
|
||||
storage.setItem(key,JSON.stringify(conflicted));
|
||||
adopt(currentWeek);
|
||||
return {today,week:currentWeek,conflict:true,sync_pending:true};
|
||||
if(!pendingPull()||!pullKey())return Promise.resolve(false);
|
||||
const run=async()=>{
|
||||
let result,lastResult=false;
|
||||
while(pendingPull()){
|
||||
const current=pendingPull();
|
||||
try {
|
||||
result=await fetchJson('api/v1/week/pull-item',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(current.body)});
|
||||
} catch(error) {
|
||||
if(error?.status!==409)throw error;
|
||||
const [today,currentWeek]=await Promise.all([fetchJson('api/v1/today'),fetchJson('api/v1/week')]);
|
||||
const queue=pullQueue(),conflicted={...current,conflict:true,current:{today,week:currentWeek}};
|
||||
if(!writePullQueue([conflicted,...queue.slice(1)]))throw new Error('Could not preserve queued Week Ahead work on this device.');
|
||||
adopt(currentWeek);
|
||||
return {today,week:currentWeek,conflict:true,sync_pending:true};
|
||||
}
|
||||
const queue=pullQueue();
|
||||
if(queue[0]?.body?.operation_id!==current.body.operation_id)continue;
|
||||
const remaining=queue.slice(1).map(entry=>{
|
||||
const identity=entry.body.identity;
|
||||
const source=entry.week.days.find(day=>(day.ids||[]).includes(identity));
|
||||
const estimate=Number(entry.today?.estimates?.[identity]??source?.estimates?.[identity]);
|
||||
const today={...result.today,ids:(result.today.ids||[]).includes(identity)?[...result.today.ids]:[...result.today.ids,identity],
|
||||
estimates:{...(result.today.estimates||{}),...(Number.isFinite(estimate)?{[identity]:estimate}:{})}};
|
||||
const nextWeek={revision:result.week.revision,timezone:result.week.timezone||null,days:result.week.days.map(day=>{
|
||||
const estimates={...(day.estimates||{})};delete estimates[identity];
|
||||
return {...cloneDay(day),ids:(day.ids||[]).filter(id=>id!==identity),estimates};
|
||||
})};
|
||||
result={today,week:nextWeek};
|
||||
return {...entry,body:{...entry.body,today_revision:today.revision,week_revision:nextWeek.revision},today,week:nextWeek};
|
||||
});
|
||||
if(!writePullQueue(remaining))throw new Error('Could not update queued Week Ahead work on this device.');
|
||||
lastResult=result;
|
||||
}
|
||||
const current=pendingPull();
|
||||
if(current?.body?.operation_id===record.body.operation_id)storage.removeItem(key);
|
||||
adoptConfirmed(result.week,{...confirmedItems,...pendingItems});
|
||||
return {...result,sync_pending:Boolean(pendingPull())};
|
||||
})().finally(()=>{pulling=null;});
|
||||
adoptConfirmed(lastResult.week,{...confirmedItems,...pendingItems});
|
||||
return {...lastResult,sync_pending:false};
|
||||
};
|
||||
const delivery=coordinator?coordinator.runExclusive('week',run):run();
|
||||
pulling=Promise.resolve(delivery).finally(()=>{pulling=null;});
|
||||
return pulling;
|
||||
}
|
||||
async function startEarly(planDate,todayRevision) {
|
||||
|
|
|
|||
|
|
@ -2,14 +2,18 @@ import json
|
|||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
FRONTEND = Path(__file__).parents[1] / "frontend"
|
||||
CONTROLLER = FRONTEND / "week-plan.js"
|
||||
COORDINATOR = FRONTEND / "outbox-coordinator.js"
|
||||
|
||||
|
||||
def run_controller(scenario: str) -> dict:
|
||||
harness = f"""
|
||||
const createWeekPlan = require({json.dumps(str(CONTROLLER))});
|
||||
const createOutboxCoordinator = require({json.dumps(str(COORDINATOR))});
|
||||
(async()=>{{ {scenario} }})().catch(error=>{{console.error(error);process.exit(1);}});
|
||||
"""
|
||||
completed = subprocess.run(["node", "-e", harness], check=True, capture_output=True, text=True)
|
||||
|
|
@ -358,6 +362,128 @@ console.log(JSON.stringify({pending,resume,confirmed,requests,keys:[...values.ke
|
|||
assert result["state"]["revision"] == 8
|
||||
|
||||
|
||||
def test_week_controller_queues_two_offline_pulls_and_replays_them_fifo_after_reload():
|
||||
result = run_controller("""
|
||||
const values=new Map(),delivered=[];
|
||||
const storage={getItem:key=>values.get(key)||null,setItem:(key,value)=>values.set(key,value),removeItem:key=>values.delete(key)};
|
||||
let offline=true;
|
||||
let serverToday={revision:4,ids:['active'],capacity_minutes:180,estimates:{active:30}};
|
||||
let serverWeek={revision:7,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['first','second','later'],capacity_minutes:150,estimates:{first:35,second:40,later:45}}]};
|
||||
const fetchJson=async(url,options={})=>{
|
||||
if(offline)throw new Error('connection lost');
|
||||
const body=JSON.parse(options.body);delivered.push(body);
|
||||
if(body.today_revision!==serverToday.revision||body.week_revision!==serverWeek.revision){const error=new Error('changed');error.status=409;throw error;}
|
||||
serverToday={...serverToday,revision:serverToday.revision+1,ids:[...serverToday.ids,body.identity],
|
||||
estimates:{...serverToday.estimates,[body.identity]:serverWeek.days[0].estimates[body.identity]}};
|
||||
serverWeek={...serverWeek,revision:serverWeek.revision+1,days:serverWeek.days.map(day=>({...day,
|
||||
ids:day.ids.filter(id=>id!==body.identity),estimates:Object.fromEntries(Object.entries(day.estimates).filter(([id])=>id!==body.identity))}))};
|
||||
return {today:serverToday,week:serverWeek};
|
||||
};
|
||||
const options={storage,getLogin:()=> 'Timmy',fetchJson,localDate:()=> '2026-08-20',timeZone:()=> 'UTC'};
|
||||
const first=createWeekPlan(options);first.adopt(serverWeek);
|
||||
const queuedFirst=await first.pullItem('first',serverToday,'pull-first');
|
||||
const queuedSecond=await first.pullItem('second',queuedFirst.today,'pull-second');
|
||||
const restored=createWeekPlan(options),resumed=restored.resumePull();
|
||||
offline=false;
|
||||
const confirmed=await restored.flushPull();
|
||||
console.log(JSON.stringify({queuedFirst,queuedSecond,resumed,confirmed,delivered,keys:[...values.keys()],state:restored.state()}));
|
||||
""")
|
||||
|
||||
assert result["queuedFirst"]["today"]["ids"] == ["active", "first"]
|
||||
assert result["queuedSecond"]["today"]["ids"] == ["active", "first", "second"]
|
||||
assert result["resumed"]["today"]["ids"] == ["active", "first", "second"]
|
||||
assert [request["operation_id"] for request in result["delivered"]] == ["pull-first", "pull-second"]
|
||||
assert result["delivered"][1]["today_revision"] == 5
|
||||
assert result["delivered"][1]["week_revision"] == 8
|
||||
assert result["confirmed"]["today"]["ids"] == ["active", "first", "second"]
|
||||
assert result["confirmed"]["week"]["days"][0]["ids"] == ["later"]
|
||||
assert result["confirmed"]["sync_pending"] is False
|
||||
assert result["keys"] == ["stackchain.week-confirmed.v1.timmy"]
|
||||
assert result["state"]["revision"] == 9
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"coordinator_setup",
|
||||
[
|
||||
"""const locks={request:async(name,_options,work)=>{
|
||||
if(held.has(name))return work(null);
|
||||
held.add(name);try{return await work({name});}finally{held.delete(name);}
|
||||
}};
|
||||
const coordinatorA=createOutboxCoordinator({storage,locks,channelFactory:null,tabId:'a'});
|
||||
const coordinatorB=createOutboxCoordinator({storage,locks,channelFactory:null,tabId:'b'});""",
|
||||
"""const coordinatorA=createOutboxCoordinator({storage,locks:null,channelFactory:null,tabId:'a',now:()=>100});
|
||||
const coordinatorB=createOutboxCoordinator({storage,locks:null,channelFactory:null,tabId:'b',now:()=>100});""",
|
||||
],
|
||||
ids=["navigator-lock", "storage-lease"],
|
||||
)
|
||||
def test_week_controller_allows_only_one_tab_to_replay_the_shared_pull_queue(coordinator_setup):
|
||||
result = run_controller("""
|
||||
const values=new Map(),held=new Set();
|
||||
const storage={getItem:key=>values.get(key)||null,setItem:(key,value)=>values.set(key,value),removeItem:key=>values.delete(key)};
|
||||
COORDINATOR_SETUP
|
||||
let offline=true,posts=0,release;
|
||||
const gate=new Promise(resolve=>release=resolve);
|
||||
const fetchJson=async(_url,options={})=>{
|
||||
if(offline)throw new Error('connection lost');
|
||||
posts+=1;await gate;
|
||||
const body=JSON.parse(options.body);
|
||||
return {today:{revision:5,ids:['active',body.identity],capacity_minutes:180,estimates:{active:30,[body.identity]:35}},
|
||||
week:{revision:8,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:[],capacity_minutes:150,estimates:{}}]}};
|
||||
};
|
||||
const options={storage,getLogin:()=> 'timmy',fetchJson,localDate:()=> '2026-08-20',timeZone:()=> 'UTC'};
|
||||
const source=createWeekPlan({...options,coordinator:coordinatorA});source.adopt({revision:7,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['first'],capacity_minutes:150,estimates:{first:35}}]});
|
||||
await source.pullItem('first',{revision:4,ids:['active'],capacity_minutes:180,estimates:{active:30}},'pull-first');
|
||||
offline=false;
|
||||
const first=createWeekPlan({...options,coordinator:coordinatorA}),second=createWeekPlan({...options,coordinator:coordinatorB});
|
||||
const firstFlush=first.flushPull();
|
||||
await Promise.resolve();
|
||||
setImmediate(release);
|
||||
const skipped=await second.flushPull();
|
||||
release();
|
||||
const confirmed=await firstFlush;
|
||||
console.log(JSON.stringify({posts,skipped,confirmed,pending:first.pendingPull()}));
|
||||
""".replace("COORDINATOR_SETUP", coordinator_setup))
|
||||
|
||||
assert result["posts"] == 1
|
||||
assert result["skipped"]["lease_skipped"] is True
|
||||
assert result["confirmed"]["sync_pending"] is False
|
||||
assert result["pending"] is None
|
||||
|
||||
|
||||
def test_week_controller_resumes_server_truth_when_a_fifo_head_conflicts():
|
||||
result = run_controller("""
|
||||
const values=new Map();
|
||||
const storage={getItem:key=>values.get(key)||null,setItem:(key,value)=>values.set(key,value),removeItem:key=>values.delete(key)};
|
||||
let offline=true,posts=0;
|
||||
const remoteToday={revision:10,ids:['server-today'],capacity_minutes:180,estimates:{'server-today':30}};
|
||||
const remoteWeek={revision:12,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['second','third','server-week'],capacity_minutes:180,estimates:{second:40,third:45,'server-week':25}}]};
|
||||
const fetchJson=async(url,options={})=>{
|
||||
if(offline)throw new Error('connection lost');
|
||||
if(!options.method)return url.endsWith('/today')?remoteToday:remoteWeek;
|
||||
posts+=1;
|
||||
if(posts===1)return {today:{revision:5,ids:['active','first'],capacity_minutes:180,estimates:{active:30,first:35}},
|
||||
week:{revision:8,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['second','third'],capacity_minutes:180,estimates:{second:40,third:45}}]}};
|
||||
const error=new Error('changed');error.status=409;throw error;
|
||||
};
|
||||
const options={storage,getLogin:()=> 'timmy',fetchJson,localDate:()=> '2026-08-20',timeZone:()=> 'UTC'};
|
||||
const first=createWeekPlan(options);first.adopt({revision:7,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['first','second','third'],capacity_minutes:180,estimates:{first:35,second:40,third:45}}]});
|
||||
let today={revision:4,ids:['active'],capacity_minutes:180,estimates:{active:30}};
|
||||
today=(await first.pullItem('first',today,'pull-first')).today;
|
||||
today=(await first.pullItem('second',today,'pull-second')).today;
|
||||
await first.pullItem('third',today,'pull-third');
|
||||
offline=false;const conflicted=await first.flushPull();
|
||||
const restored=createWeekPlan(options),resumed=restored.resumePull();
|
||||
console.log(JSON.stringify({conflicted,resumed,state:restored.state(),pending:restored.pendingPull()}));
|
||||
""")
|
||||
|
||||
assert result["conflicted"]["conflict"] is True
|
||||
assert result["resumed"]["conflict"] is True
|
||||
assert result["resumed"]["today"]["ids"] == ["server-today"]
|
||||
assert result["state"]["revision"] == 12
|
||||
assert result["state"]["days"][0]["ids"] == ["second", "third", "server-week"]
|
||||
assert result["pending"]["body"]["operation_id"] == "pull-second"
|
||||
|
||||
|
||||
def test_week_controller_preserves_a_conflicted_pull_intent_and_adopts_server_truth():
|
||||
result = run_controller("""
|
||||
const values=new Map(),requests=[];
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user