Speed up direct sync finance reconciliation

scripts/cloud/legacy-finance.js

@@ -28,26 +28,37 @@ async function importLegacyFinance(co,uid,runId,readSource){
   if(all.length!==total)throw Error('finance_source_count_mismatch');
   const result=await Psql.tx(async t=>{
    await t.any("SELECT pg_advisory_xact_lock(hashtext('xs-credit:'||$1))",[co]);
-   let created=0;const ids=[];
-   for(const raw of all){
+   let created=0;const ids=[],patches=[],inserts=[];
+   const sourceIds=all.map(raw=>String(raw.ExpHisID));
+   const hashes=await t.any('SELECT ord::int idx,md5(value::text) hash FROM jsonb_array_elements($1::jsonb) WITH ORDINALITY AS e(value,ord)',[JSON.stringify(all)]);
+   if(hashes.length!==all.length)throw Error('finance_hash_count_mismatch');
+   const existing=sourceIds.length?await t.any('SELECT to_jsonb(l) j FROM "'+table+'" l WHERE company=$1 AND ("legacyFinanceId"=ANY($2::text[]) OR "sourceKey"=ANY($3::text[]) OR to_jsonb(l)->>\'expHisId\'=ANY($2::text[]))',[co,sourceIds,sourceIds.map(id=>'[["ExpHisID",'+id+']]')]):[];
+   const priorById=new Map(),sourceIdSet=new Set(sourceIds);
+   for(const {j} of existing){
+    const keys=[j.legacyFinanceId,j.expHisId,/^\[\["ExpHisID",(\d+)\]\]$/.exec(String(j.sourceKey||''))?.[1]].filter(Boolean).map(String);
+    for(const key of new Set(keys))if(sourceIdSet.has(key)){const bucket=priorById.get(key)||[];bucket.push(j);priorById.set(key,bucket);}
+   }
+   for(let index=0;index<all.length;index++){
+    const raw=all[index];
     const id=String(raw.ExpHisID),delta=Number(raw.score??raw.Score),before=raw.score_before===null||raw.score_before===undefined||raw.score_before===''?null:Number(raw.score_before);
     const date=String(raw.HisTime||'').replace(' ','T'),iso=/([+-]\d{2}:?\d{2}|Z)$/.test(date)?date:date+'+08:00';
     if((raw.score??raw.Score)===undefined||(raw.score??raw.Score)===null||!Number.isFinite(delta)||(before!==null&&!Number.isFinite(before))||Number.isNaN(Date.parse(iso)))throw Error('finance_invalid_values');
-    const hash=(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(raw)])).hash;
-    const prior=await t.any('SELECT to_jsonb(l) j FROM "'+table+'" l WHERE company=$1 AND ("legacyFinanceId"=$2 OR "sourceKey"=$3 OR to_jsonb(l)->>\'expHisId\'=$2)',[co,id,'[["ExpHisID",'+id+']]']);
+    const hash=hashes[index].hash,prior=(priorById.get(id)||[]);
     if(prior.length>1)throw Error('finance_duplicate_source');
-    if(prior.length){const j=prior[0].j;if(j.legacyFinanceHash&&j.legacyFinanceHash!==hash)throw Error('finance_source_conflict_preserved');if(Number(j.userId)!==uid||Number(j.score)!==delta||((j.scoreBefore===null?null:Number(j.scoreBefore))!==before))throw Error('finance_existing_row_conflict');
+    if(prior.length){const j=prior[0];if(j.legacyFinanceId&&String(j.legacyFinanceId)!==id)throw Error('finance_existing_row_conflict');if(j.legacyFinanceHash&&j.legacyFinanceHash!==hash)throw Error('finance_source_conflict_preserved');if(Number(j.userId)!==uid||Number(j.score)!==delta||((j.scoreBefore===null?null:Number(j.scoreBefore))!==before))throw Error('finance_existing_row_conflict');
      if(Date.parse(j.hisTime)!==Date.parse(iso)){
       if(!untouchedLegacyFinanceTimezoneRow(j,raw,id,iso))throw Error('finance_existing_row_conflict');
       const corrected=new Date(iso).toISOString();
       await t.none('INSERT INTO "LegacySyncConflict"("objectId",company,"sourceKey",dataset,"legacyKey","conflictType",message,resolved,resolution,"createdAt","updatedAt") VALUES(xs_object_id(),$1,$2,\'finance\',$3,\'finance-legacy-timezone-repair\',$4,true,\'原始流水核验一致，仅修复早期导入时区偏移\',now(),now())',[co,'finance-timezone:'+account+':'+id,id,JSON.stringify({runId,account,studentId:uid,before:j,after:{...j,hisTime:corrected},sourceTime:raw.HisTime})]);
       await t.none('UPDATE "'+table+'" SET "hisTime"=$2::timestamptz WHERE "objectId"=$1',[j.objectId,corrected]);
      }
-     await t.none('UPDATE "'+table+'" SET "legacyFinanceId"=$2,"legacyFinanceHash"=$3,"legacyFinanceRaw"=$4::jsonb WHERE "objectId"=$1',[j.objectId,id,hash,JSON.stringify(raw)]);
+     if(j.legacyFinanceId!==id||j.legacyFinanceHash!==hash||!j.legacyFinanceRaw)patches.push({oid:j.objectId,id,hash,raw});
     }else{
-     await t.none('INSERT INTO "'+table+'" SELECT * FROM jsonb_populate_record(NULL::"'+table+'",$1::jsonb)',[JSON.stringify({objectId:(await t.one('SELECT xs_object_id() id')).id,company:co,sourceKey:'legacy:finance:'+account+':'+id,userId:uid,score:delta,scoreBefore:before,hisTime:new Date(iso).toISOString(),createdAt:new Date().toISOString(),updatedAt:new Date().toISOString(),detail:String(raw.Detail||''),remark:String(raw.Remark||''),operator:Number(raw.Operator)||0,scoreType:Number(raw.ScoreType)||0,legacyFinanceId:id,legacyFinanceRaw:raw,legacyFinanceHash:hash})]);created++;
+     inserts.push({company:co,sourceKey:'legacy:finance:'+account+':'+id,userId:uid,score:delta,scoreBefore:before,hisTime:new Date(iso).toISOString(),createdAt:new Date().toISOString(),updatedAt:new Date().toISOString(),detail:String(raw.Detail||''),remark:String(raw.Remark||''),operator:Number(raw.Operator)||0,scoreType:Number(raw.ScoreType)||0,legacyFinanceId:id,legacyFinanceRaw:raw,legacyFinanceHash:hash});created++;
     }ids.push(id);
    }
+   for(let i=0;i<patches.length;i+=200)await t.none('UPDATE "'+table+'" l SET "legacyFinanceId"=x.id,"legacyFinanceHash"=x.hash,"legacyFinanceRaw"=x.raw FROM jsonb_to_recordset($1::jsonb) AS x(oid text,id text,hash text,raw jsonb) WHERE l."objectId"=x.oid',[JSON.stringify(patches.slice(i,i+200))]);
+   for(let i=0;i<inserts.length;i+=200){const rows=inserts.slice(i,i+200),objectIds=await t.any('SELECT xs_object_id() id FROM generate_series(1,$1::int)',[rows.length]);if(objectIds.length!==rows.length)throw Error('finance_object_id_allocation_failed');for(let j=0;j<rows.length;j++)rows[j].objectId=objectIds[j].id;await t.none('INSERT INTO "'+table+'" SELECT * FROM jsonb_populate_recordset(NULL::"'+table+'",$1::jsonb)',[JSON.stringify(rows)]);}
    ids.sort((a,b)=>Number(a)-Number(b));
    const actual=await t.any('SELECT "legacyFinanceId" id,"legacyFinanceHash" hash FROM "'+table+'" WHERE company=$1 AND "userId"=$2 AND "legacyFinanceId"=ANY($3::text[]) ORDER BY "legacyFinanceId"::numeric',[co,uid,ids]);
    if(JSON.stringify(actual.map(r=>r.id))!==JSON.stringify(ids))throw Error('finance_id_reconciliation_failed');

scripts/run-nightly-direct-sync.mjs

@@ -115,7 +115,15 @@ export async function runDirectSync(env=process.env,args=process.argv.slice(2)){
    }
   }
   report.finance=report.finance||{members:[],created:0};
-  for(const member of await call({action:'finance-members'},2)){const result=await call({action:'finance',studentId:Number(member.id)},2);report.finance.members.push(result);report.finance.created+=result.created;await save(join(root,'direct-'+runId,'finance.json'),report.finance);}
+  const financeMembers=await call({action:'finance-members'},2);
+  // Bound source requests while server-side account transactions retain their lock.
+  // Wait for every request in a batch before releasing the run lock on failure.
+  for(let i=0;i<financeMembers.length;i+=4){
+   const batch=await Promise.allSettled(financeMembers.slice(i,i+4).map(member=>call({action:'finance',studentId:Number(member.id)},2)));
+   for(const item of batch)if(item.status==='fulfilled'){report.finance.members.push(item.value);report.finance.created+=item.value.created;}
+   await save(join(root,'direct-'+runId,'finance.json'),report.finance);
+   const failed=batch.find(item=>item.status==='rejected');if(failed)throw failed.reason;
+  }
   report.createdBusinessRecords=report.datasets.reduce((n,d)=>n+d.created,0);report.refreshedBusinessRecords=report.datasets.reduce((n,d)=>n+d.updated,0);report.accounts=report.supplement.accounts;report.nodes=report.supplement.nodes;report.conflicts=report.datasets.reduce((n,d)=>n+d.conflicts,0)+Number(report.accounts.conflicts||0);
   report.status=report.conflicts?'completed-with-preserved-conflicts':'completed';report.allCapturedSourceIdsReconciled=true;
   Object.assign(report,await call({action:'finish',summary:report},2));

scripts/tests/direct-nightly-sync.test.mjs

@@ -97,6 +97,28 @@ test('每日运行入口只使用直接接口、按依赖顺序并写私有报
  }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
 });
 
+test('财务流水按最多四位会员并行，全部请求结束后才释放目录锁',async()=>{
+ const root=await mkdtemp(join(tmpdir(),'direct-finance-test-')),realFetch=globalThis.fetch;
+ let active=0,peak=0,completed=0;
+ try{
+  globalThis.fetch=async(url,options)=>{const p=JSON.parse(options.body).params;let data;
+   if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false};
+   else if(p.action==='claim')data={runId:'mock-finance-run-001',businessDate:'2026-10-12'};
+   else if(p.action==='supplement')data={accounts:{missing:0},nodes:{missing:0}};
+   else if(p.action==='source')data={items:[],page:{itemCount:0,pageCount:1}};
+   else if(p.action==='read')data=[];
+   else if(p.action==='evidence')data={sourceChecksum:'d41d8cd98f00b204e9800998ecf8427e',targetChecksum:'d41d8cd98f00b204e9800998ecf8427e',reconciled:true};
+   else if(p.action==='finance-members')data=[1,2,3,4,5].map(id=>({id}));
+   else if(p.action==='finance'){active++;peak=Math.max(peak,active);await new Promise(r=>setTimeout(r,15));active--;completed++;data={studentId:p.studentId,accounts:[],created:0};}
+   else if(p.action==='finish')data=p.summary;
+   return {ok:true,json:async()=>({success:true,data})};
+  };
+  const result=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});
+  assert.equal(result.status,'completed',result.reason);assert.equal(peak,4);assert.equal(completed,5);
+  assert.equal(result.finance.members.length,5);assert.ok(!(await readdir(root)).includes('direct-nightly.lock'));
+ }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
+});
+
 
 test('有新系统计薪凭据的复习不回退、不重算工资',async()=>{
  const row={generalId:301,common:{objectId:'memory-common',status:99},addon:{objectId:'memory-addon',id:301,yhid:'20',xxjlid:'33',plid:'30',kywrq:'2026-10-12',fxzt:'0'}};