Verify email outbox recovery across real process crashes
This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {spawn,execFileSync} from 'node:child_process';
|
||||
import {mkdtempSync,rmSync,readFileSync} from 'node:fs';
|
||||
import {join} from 'node:path';
|
||||
import {tmpdir} from 'node:os';
|
||||
const storeUrl=new URL('../server/notify-store.mjs',import.meta.url).href;
|
||||
const notifierUrl=new URL('../server/notifier.mjs',import.meta.url).href;
|
||||
const shared=`
|
||||
import assert from 'node:assert/strict';
|
||||
import fs from 'node:fs';
|
||||
import {DatabaseSync} from 'node:sqlite';
|
||||
import * as store from ${JSON.stringify(storeUrl)};
|
||||
import {bildirimleriIsle} from ${JSON.stringify(notifierUrl)};
|
||||
const providerFile=process.env.IHALEIST_USER_DATA_DIR+'/synthetic-provider.json';
|
||||
// A deterministic local provider simulator; never sends a network request.
|
||||
const provider=async message=>{
|
||||
const state=fs.existsSync(providerFile)?JSON.parse(fs.readFileSync(providerFile,'utf8')):{requests:0,deliveries:0,records:{}};
|
||||
state.requests++;
|
||||
const key=message.idempotencyKey,body=JSON.stringify(message);
|
||||
assert.ok(key);
|
||||
if(state.records[key])assert.equal(state.records[key].body,body,'retry payload must be byte-identical');
|
||||
else {state.deliveries++;state.records[key]={body,id:'synthetic-provider-receipt'};}
|
||||
fs.writeFileSync(providerFile,JSON.stringify(state));return {id:state.records[key].id};
|
||||
};
|
||||
const tender={id:'outbox-tender',ikn:'2026/9',baslik:'Laboratuvar source A',tarih:'10.09.2026',saat:'10:00',durum:'Açık'};
|
||||
const options={now:()=>new Date('2026-09-06T00:00:00Z'),configured:()=>true,telegramReady:()=>false,deliverEmail:provider};
|
||||
`;
|
||||
async function killAtCheckpoint(dir,phase){
|
||||
const code=shared+`
|
||||
const sub=store.subscribe('crash@example.test',{konular:['Laboratuvar']})[0];store.markBaseline(sub.id,[]);
|
||||
await bildirimleriIsle(async()=>({tenders:[tender]}),{},{...options,accepted:(delivery,providerId)=>{
|
||||
if(${JSON.stringify(phase)}==='after-commit')store.acceptDelivery(delivery,providerId);
|
||||
process.send({id:delivery.id,subscriptionId:sub.id});
|
||||
Atomics.wait(new Int32Array(new SharedArrayBuffer(4)),0,0);
|
||||
}});
|
||||
`;
|
||||
const child=spawn(process.execPath,['--input-type=module','-e',code],{env:{...process.env,IHALEIST_USER_DATA_DIR:dir,IHALEIST_SHARED_DATA_DIR:join(dir,'shared')},stdio:['ignore','ignore','pipe','ipc']});
|
||||
let errors='',timer;child.stderr.on('data',chunk=>errors+=chunk);
|
||||
try {
|
||||
const checkpoint=await new Promise((resolve,reject)=>{
|
||||
child.once('message',resolve);child.once('error',reject);child.once('exit',()=>reject(new Error('Worker exited before checkpoint: '+errors)));
|
||||
timer=setTimeout(()=>reject(new Error('Worker checkpoint timeout: '+errors)),10000);
|
||||
});
|
||||
const exit=new Promise(resolve=>child.once('exit',(code,signal)=>resolve(signal)));
|
||||
child.kill('SIGKILL');assert.equal(await exit,'SIGKILL');return checkpoint;
|
||||
}finally{clearTimeout(timer);if(child.exitCode===null&&child.signalCode===null)child.kill('SIGKILL');}
|
||||
}
|
||||
for(const phase of ['before-commit','after-commit'])test(`outbox survives SIGKILL ${phase} without a second synthetic delivery`,async()=>{
|
||||
const dir=mkdtempSync(join(tmpdir(),'ihaleist-outbox-crash-'));
|
||||
try {
|
||||
const checkpoint=await killAtCheckpoint(dir,phase);
|
||||
const code=shared+`
|
||||
const db=new DatabaseSync(process.env.IHALEIST_USER_DATA_DIR+'/notify.sqlite');
|
||||
const initial=db.prepare('SELECT state FROM notification_outbox WHERE id=?').get(${JSON.stringify(checkpoint.id)}).state;db.close();
|
||||
const result=await bildirimleriIsle(async()=>({tenders:[{...tender,baslik:'Laboratuvar source B'}]}),{},{...options,reserve:(id,payload,ids,settings)=>store.reserveDelivery(id,payload,ids,{...settings,clock:Date.now()+121000})});
|
||||
const saved=store.list('crash@example.test')[0];
|
||||
const check=new DatabaseSync(process.env.IHALEIST_USER_DATA_DIR+'/notify.sqlite');
|
||||
const final=check.prepare('SELECT state,provider_id FROM notification_outbox WHERE id=?').get(${JSON.stringify(checkpoint.id)});
|
||||
const integrity=check.prepare('PRAGMA integrity_check').get().integrity_check;check.close();
|
||||
fs.writeFileSync(process.env.IHALEIST_USER_DATA_DIR+'/result.json',JSON.stringify({initial,result,saved,final,integrity}));
|
||||
`;
|
||||
execFileSync(process.execPath,['--input-type=module','-e',code],{env:{...process.env,IHALEIST_USER_DATA_DIR:dir,IHALEIST_SHARED_DATA_DIR:join(dir,'shared')},timeout:10000,stdio:'pipe'});
|
||||
const recovered=JSON.parse(readFileSync(join(dir,'result.json'),'utf8'));
|
||||
const provider=JSON.parse(readFileSync(join(dir,'synthetic-provider.json'),'utf8'));
|
||||
assert.equal(recovered.initial,phase==='before-commit'?'sending':'sent');
|
||||
assert.equal(recovered.final.state,'sent');assert.equal(recovered.final.provider_id,'synthetic-provider-receipt');assert.equal(recovered.integrity,'ok');
|
||||
assert.deepEqual(recovered.saved.sonGorulenler,['outbox-tender']);
|
||||
assert.equal(provider.deliveries,1);assert.equal(provider.requests,phase==='before-commit'?2:1);
|
||||
}finally{rmSync(dir,{recursive:true,force:true});}
|
||||
});
|
||||
Reference in New Issue
Block a user