feat: make background queue admission durable (#264)
This commit is contained in:
1 parent
94aad3e76e
commit
c74c63d29c
6 files changed
+194
-28
No files matched your search
@@ -182,6 +182,35 @@ setTimeout(() => {{
|
||||
]
|
||||
|
||||
|
||||
def test_authored_outbox_waits_for_durable_background_admission():
|
||||
script = f"""
|
||||
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
|
||||
const values = new Map();
|
||||
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
|
||||
let release; const gate = new Promise(resolve => release = resolve); const events = [];
|
||||
const outbox = createAuthoredOutbox({{
|
||||
storage, getOwnerLogin:()=> 'timmy',
|
||||
backgroundSync: {{
|
||||
reconcile: async (_items, lane) => {{ events.push('mirror:' + lane); await gate; events.push('committed'); }},
|
||||
requestSync: async () => events.push('registered'),
|
||||
}},
|
||||
}});
|
||||
const admission = outbox.enqueueDurably({{
|
||||
kind:'issue-comment',repository:'o/r',number:1,body:'Do not lose',operationId:'message-1'
|
||||
}}).then(result => {{ events.push('confirmed'); return result; }});
|
||||
Promise.resolve().then(async () => {{
|
||||
const pending = events.slice(); release(); const result = await admission;
|
||||
process.stdout.write(JSON.stringify({{pending,events,result,items:outbox.list()}}));
|
||||
}});
|
||||
"""
|
||||
output = run_node(script)
|
||||
|
||||
assert output["pending"] == ["mirror:authored"]
|
||||
assert output["events"] == ["mirror:authored", "committed", "registered", "confirmed"]
|
||||
assert output["result"]["durability"] == "background"
|
||||
assert output["items"][0]["operationId"] == "message-1"
|
||||
|
||||
|
||||
def test_authored_outbox_foreground_send_uses_atomic_background_delivery():
|
||||
script = f"""
|
||||
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
|
||||
@@ -217,7 +246,9 @@ async def test_mobile_dashboard_loads_and_operates_authored_message_outbox():
|
||||
assert "const authoredOutbox = createAuthoredOutbox({" in html
|
||||
assert "activeFlushLogin = contextIdentityFresh ?" in html
|
||||
assert "flushAuthoredOutbox();" in html
|
||||
assert "authoredOutbox.enqueue" in html
|
||||
assert "await authoredOutbox.enqueueDurably" in html
|
||||
assert "Saving for background delivery…" in html
|
||||
assert "Saved for next launch; background delivery unavailable." in html
|
||||
assert "authoredOutbox.retry(item.outbox_id, activeFlushLogin)" in html
|
||||
assert "authoredOutbox.discard(item.outbox_id)" in html
|
||||
assert "if (result?.queued)" in html
|
||||
|
||||
@@ -186,6 +186,70 @@ setTimeout(() => process.stdout.write(JSON.stringify(state)), 0);
|
||||
assert output["syncs"] == 1
|
||||
|
||||
|
||||
def test_issue_outbox_confirms_durable_admission_only_after_mirror_and_sync():
|
||||
script = f"""
|
||||
const createIssueOutbox = require({json.dumps(str(OUTBOX))});
|
||||
const values = new Map();
|
||||
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
|
||||
let releaseMirror; let releaseSync;
|
||||
const mirrorGate = new Promise(resolve => releaseMirror = resolve);
|
||||
const syncGate = new Promise(resolve => releaseSync = resolve);
|
||||
const events = [];
|
||||
const outbox = createIssueOutbox({{
|
||||
storage, getOwnerLogin:()=> 'timmy', createOperationId:()=> 'durable-1',
|
||||
backgroundSync: {{
|
||||
reconcile: async items => {{ events.push('mirror-start'); await mirrorGate; events.push('mirror-committed'); }},
|
||||
requestSync: async () => {{ events.push('sync-start'); await syncGate; events.push('sync-registered'); }},
|
||||
}},
|
||||
}});
|
||||
const admission = outbox.enqueueDurably({{repository:'stackchain/api',title:'Keep this',body:'Draft'}})
|
||||
.then(result => {{ events.push('confirmed'); return result; }});
|
||||
Promise.resolve().then(async () => {{
|
||||
const beforeMirror = events.slice();
|
||||
releaseMirror(); await new Promise(resolve => setTimeout(resolve, 0));
|
||||
const beforeSync = events.slice();
|
||||
releaseSync();
|
||||
const result = await admission;
|
||||
process.stdout.write(JSON.stringify({{beforeMirror,beforeSync,events,result,items:outbox.list()}}));
|
||||
}});
|
||||
"""
|
||||
output = run_node(script)
|
||||
|
||||
assert output["beforeMirror"] == ["mirror-start"]
|
||||
assert output["beforeSync"] == ["mirror-start", "mirror-committed", "sync-start"]
|
||||
assert output["events"][-1] == "confirmed"
|
||||
assert output["result"]["background"] is True
|
||||
assert output["result"]["item"]["id"] == "durable-1"
|
||||
assert output["items"][0]["title"] == "Keep this"
|
||||
|
||||
|
||||
def test_issue_outbox_reports_degraded_admission_without_losing_foreground_item():
|
||||
script = f"""
|
||||
const createIssueOutbox = require({json.dumps(str(OUTBOX))});
|
||||
const values = new Map();
|
||||
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
|
||||
const outbox = createIssueOutbox({{
|
||||
storage, getOwnerLogin:()=> 'timmy', createOperationId:()=> 'foreground-1',
|
||||
backgroundSync: {{
|
||||
reconcile: async () => {{}},
|
||||
requestSync: async () => {{ throw new Error('Background Sync unavailable'); }},
|
||||
}},
|
||||
}});
|
||||
outbox.enqueueDurably({{repository:'stackchain/api',title:'Retain me',body:'Draft'}}).then(result =>
|
||||
process.stdout.write(JSON.stringify({{
|
||||
background:result.background, durability:result.durability,
|
||||
error:result.error.message, item:result.item, persisted:outbox.list()
|
||||
}}))
|
||||
);
|
||||
"""
|
||||
output = run_node(script)
|
||||
|
||||
assert output["background"] is False
|
||||
assert output["durability"] == "foreground-only"
|
||||
assert output["error"] == "Background Sync unavailable"
|
||||
assert output["persisted"] == [output["item"]]
|
||||
|
||||
|
||||
def test_foreground_delivery_uses_same_atomic_background_claim():
|
||||
script = f"""
|
||||
const createIssueOutbox = require({json.dumps(str(OUTBOX))});
|
||||
@@ -263,7 +327,10 @@ async def test_mobile_dashboard_queues_offline_captures_and_exposes_outbox_actio
|
||||
|
||||
assert '<script src="static/issue-outbox.js"></script>' in html
|
||||
assert "const issueOutbox = createIssueOutbox({" in html
|
||||
assert "issueOutbox.enqueue(captureDraft)" in html
|
||||
assert "await issueOutbox.enqueueDurably(captureDraft)" in html
|
||||
assert "Saving for background delivery…" in html
|
||||
assert "Saved for next launch; background delivery unavailable." in html
|
||||
assert "throw new Error('Background Sync unavailable')" in html
|
||||
assert "issueOutbox.retry(queued.id, activeFlushLogin)" in html
|
||||
assert "navigator.onLine" in html
|
||||
assert "activeFlushLogin = contextIdentityFresh ?" in html
|
||||
|
||||
Reference in new issue
Block a user