// Manual acceptance helper for a PRIVATE cloned schema. Import into the existing // fixed diagnostic gateway harness; this module opens no connection or remote run. // Pass real SQL scripts as strings. The clone must already contain the native // source tables, original payroll functions, and financial rule/reward tables. export async function installPayrollSourceIndexConcurrency(Psql, config) { const {schema,incrementalSql,indexSql}=config; if(!/^speed3_[a-z0-9_]+$/.test(schema))throw Error('Private speed3_ clone schema required'); const rewrite=sql=>sql.replace(/^BEGIN;\s*$/m,'').replace(/^COMMIT;\s*$/m,'').replaceAll("n.nspname='public'",`n.nspname='${schema}'`).replaceAll('public.',schema+'.'); return Psql.tx(async tx=>{ await tx.none(`SET LOCAL search_path=${schema},public`); await tx.none("SET LOCAL statement_timeout='120s';SET LOCAL jit=off;SET LOCAL work_mem='64MB';SET LOCAL TimeZone='Etc/GMT-8';SET LOCAL extra_float_digits=1"); // CREATE TABLE AS clones do not retain the digest primary key. The real // round-2 row trigger requires uniqueness for its atomic replacement. await tx.none('CREATE UNIQUE INDEX IF NOT EXISTS concurrency_digest_identity ON "PayrollReadDigest"(kind,company,"objectId")'); await tx.any(rewrite(incrementalSql)); await tx.any(rewrite(indexSql)); const readiness=await tx.one('SELECT xs_payroll_authority_ready() authority,xs_payroll_digest_ready() digest,xs_payroll_source_index_ready() projection'); if(!readiness.authority||!readiness.digest||!readiness.projection)throw Error('Real authority/digest/projection guards are not ready'); return readiness; }); } // Self-contained so an existing diagnostic gateway can serialize this function. // Runs two REAL concurrent SQL transactions. A source writer which loses the // target advisory lock must commit without waiting for projection upkeep. Its // newer committed source is then read correctly despite the older index writer // committing last. The original and indexed reads share one RR snapshot. export async function payrollSourceIndexConcurrentAcceptance(Psql,config) { const {schema,company}=config; if(!/^speed3_[a-z0-9_]+$/.test(schema))throw Error('Private speed3_ clone schema required'); const prepare=async(tx,isolation='READ COMMITTED')=>{ await tx.none(`SET TRANSACTION ISOLATION LEVEL ${isolation}`); await tx.none(`SET LOCAL search_path=${schema},public`); await tx.none("SET LOCAL statement_timeout='30s';SET LOCAL lock_timeout='3s';SET LOCAL jit=off;SET LOCAL work_mem='64MB';SET LOCAL TimeZone='Etc/GMT-8';SET LOCAL extra_float_digits=1"); }; const fixture=await Psql.tx(async tx=>{ await prepare(tx); const guards=await tx.one(`SELECT xs_payroll_source_index_ready() ready,(SELECT count(*)::int FROM pg_trigger WHERE tgrelid IN ('"CommonModel"'::regclass,'"CourseAppointment"'::regclass,'"LessonRecord"'::regclass,'"MemoryPracticeRecord"'::regclass) AND tgname='xs_payroll_digest_row' AND tgenabled='A') triggers`); if(!guards.ready||guards.triggers!==4)throw Error('Real digest triggers must be installed; static profile stubs are insufficient'); const found=await tx.oneOrNone(`SELECT l."objectId" lesson,l.pldp comment,a."objectId" appointment,a."durationMinutes" minutes FROM "LessonRecord" l JOIN LATERAL (SELECT a.* FROM "CourseAppointment" a JOIN "CommonModel" c ON c.company=a.company AND c."itemId"=a.id AND c."modelId"=54 WHERE a.company=l.company AND c."generalId"::text=l.yyds::text ORDER BY c."objectId" LIMIT 1) a ON true WHERE l.company=$1 ORDER BY l."objectId" LIMIT 1`,[company]); if(!found)throw Error('No real linked lesson/appointment fixture'); return found; }); const parity=async()=>Psql.tx(async tx=>{ await prepare(tx,'REPEATABLE READ'); await tx.none('SET TRANSACTION READ ONLY'); const value=await tx.one(`SELECT xs_payroll_source_index_ready() ready, (SELECT md5(jsonb_agg(s)::text) FROM xs_payroll_sources($1) s) original_rows, (SELECT md5(jsonb_agg(s)::text) FROM xs_payroll_sources_indexed($1) s) indexed_rows, xs_payroll_source_version($1) original_version,xs_payroll_source_version_indexed($1) indexed_version`,[company]); if(!value.ready||value.original_rows!==value.indexed_rows||value.original_version!==value.indexed_version)throw Error('Concurrent read parity failed: '+JSON.stringify(value)); return value; }); const results=[]; try { for(const isolation of ['READ COMMITTED','REPEATABLE READ']){ let announce,release; const firstReady=new Promise(resolve=>{announce=resolve;}),secondCommitted=new Promise(resolve=>{release=resolve;}); let firstError; const first=Psql.tx(async tx=>{ await prepare(tx,isolation); await tx.none('UPDATE "LessonRecord" SET pldp=COALESCE(pldp,\'\')||$3 WHERE company=$1 AND "objectId"=$2',[company,fixture.lesson,' index-race-'+isolation]); announce();await secondCommitted; }).catch(error=>{firstError=error;announce();throw error;}); // Attach a rejection handler before awaiting the independent writer. first.catch(()=>{}); await firstReady; if(firstError){release();await first;} let writer,whileFirstUncommitted; try { writer=await Psql.tx(async tx=>{ await prepare(tx); const lock=await tx.one("SELECT pg_try_advisory_xact_lock(hashtext('payroll-source-index'),hashtext($1||':lesson:'||$2)) acquired",[company,fixture.lesson]); if(lock.acquired)throw Error('First transaction did not hold the target projection lock'); const started=Date.now(); await tx.none('UPDATE "CourseAppointment" SET "durationMinutes"=COALESCE("durationMinutes",60)+1 WHERE company=$1 AND "objectId"=$2',[company,fixture.appointment]); return {targetLockSkipped:true,writeMs:Date.now()-started}; }); whileFirstUncommitted=await parity(); } finally {release();await first;} const after=await parity(); const stale=await Psql.tx(async tx=>{ await prepare(tx,'REPEATABLE READ'); return tx.one(`SELECT i.input_signature IS DISTINCT FROM p.input_signature stale FROM "PayrollSourceIndex" i CROSS JOIN LATERAL xs_payroll_source_index_target(i.company,i.kind,i."objectId") p WHERE i.company=$1 AND i.kind='lesson' AND i."objectId"=$2`,[company,fixture.lesson]); }); if(!stale.stale)throw Error('Race failed to leave an older projection; did not exercise read-time repair'); results.push({isolation,...writer,staleProjection:true,whileFirstUncommitted,after}); } return {fixture:{lesson:fixture.lesson,appointment:fixture.appointment},results}; } finally { await Psql.tx(async tx=>{ await prepare(tx); await tx.none('UPDATE "LessonRecord" SET pldp=$3 WHERE company=$1 AND "objectId"=$2',[company,fixture.lesson,fixture.comment]); await tx.none('UPDATE "CourseAppointment" SET "durationMinutes"=$3 WHERE company=$1 AND "objectId"=$2',[company,fixture.appointment,fixture.minutes]); }); } } // Actual UPDATE costs with all original digest triggers present. Each trial is // rolled back with a savepoint; disabling only the projection trigger supplies a // paired baseline without changing source data or old digest trigger definitions. export async function payrollSourceIndexWriterCosts(Psql,config) { const {schema,company}=config; if(!/^speed3_[a-z0-9_]+$/.test(schema))throw Error('Private speed3_ clone schema required'); return Psql.tx(async tx=>{ await tx.none(`SET LOCAL search_path=${schema},public`); await tx.none("SET LOCAL statement_timeout='60s';SET LOCAL lock_timeout='3s';SET LOCAL jit=off;SET LOCAL TimeZone='Etc/GMT-8';SET LOCAL extra_float_digits=1"); const examples=await tx.any(`SELECT 'LessonRecord' AS table_name,"objectId" FROM "LessonRecord" WHERE company=$1 ORDER BY "objectId" LIMIT 3`,[company]); examples.push(...await tx.any(`SELECT 'CourseAppointment' AS table_name,a."objectId",count(*)::int linked_lessons FROM "CourseAppointment" a JOIN "CommonModel" c ON c.company=a.company AND c."modelId"=54 AND c."itemId"=a.id JOIN "LessonRecord" l ON l.company=a.company AND l.yyds::text=c."generalId"::text WHERE a.company=$1 GROUP BY a."objectId" ORDER BY count(*) DESC,a."objectId" LIMIT 3`,[company])); examples.push(...await tx.any(`SELECT DISTINCT ON ("modelId") 'CommonModel' AS table_name,"objectId","modelId"::int model FROM "CommonModel" WHERE company=$1 AND "modelId" IN(54,59,60) ORDER BY "modelId","objectId"`,[company])); const results=[]; for(const [sample,row] of examples.entries()){ for(const enabled of sample%2?[true,false]:[false,true]){ await tx.none('SAVEPOINT writer_cost'); try { if(!enabled)await tx.none(`ALTER TABLE "${row.table_name}" DISABLE TRIGGER xs_payroll_source_index_row`); const started=Date.now(); const assignment=row.table_name==='LessonRecord'?"pldp=COALESCE(pldp,'')||' writer-cost'":row.table_name==='CourseAppointment'?'"durationMinutes"=COALESCE("durationMinutes",60)+1':"title=COALESCE(title,'')||' writer-cost'"; await tx.none(`UPDATE "${row.table_name}" SET ${assignment} WHERE company=$1 AND "objectId"=$2`,[company,row.objectId]); results.push({...row,indexEnabled:enabled,ms:Date.now()-started}); } finally {await tx.none('ROLLBACK TO SAVEPOINT writer_cost');} } } return results; }); } export async function payrollSourceIndexMembershipAcceptance(Psql,config) { const {schema,company}=config; if(!/^speed3_[a-z0-9_]+$/.test(schema))throw Error('Private speed3_ clone schema required'); const prepare=async(tx,isolation='READ COMMITTED')=>{ await tx.none(`SET TRANSACTION ISOLATION LEVEL ${isolation}`); await tx.none(`SET LOCAL search_path=${schema},public`); await tx.none("SET LOCAL statement_timeout='40s';SET LOCAL lock_timeout='3s';SET LOCAL jit=off;SET LOCAL work_mem='64MB';SET LOCAL TimeZone='Etc/GMT-8';SET LOCAL extra_float_digits=1"); }; const check=async(tx,co)=>{ const r=await tx.one(`SELECT xs_payroll_source_index_ready() ready,(SELECT md5(jsonb_agg(s)::text) FROM xs_payroll_sources($1) s) original_rows,(SELECT md5(jsonb_agg(s)::text) FROM xs_payroll_sources_indexed($1) s) indexed_rows,xs_payroll_source_version($1) original_version,xs_payroll_source_version_indexed($1) indexed_version`,[co]); if(!r.ready||r.original_rows!==r.indexed_rows||r.original_version!==r.indexed_version)throw Error('Membership parity failed: '+JSON.stringify(r));return r; }; const fixture=await Psql.tx(async tx=>{ await prepare(tx); const row=await tx.oneOrNone(`SELECT l."objectId" lesson,c."objectId" common,c."generalId" general,f."objectId" fact FROM "LessonRecord" l JOIN LATERAL (SELECT * FROM "CommonModel" c WHERE c.company=l.company AND c."modelId"=59 AND c."itemId"=l.id ORDER BY c."objectId" LIMIT 1) c ON true JOIN "LessonRecordFact" f ON f.company=l.company AND f."legacyGeneralId"::text=to_jsonb(COALESCE(c."generalId",l."legacyGeneralId",0))#>>'{}' AND f."legacyGeneralId">0 WHERE l.company=$1 ORDER BY l."objectId" LIMIT 1`,[company]); if(!row)throw Error('No canonical-suppressed fact fixture'); const next=await tx.one(`SELECT (max("generalId")+1000000)::text value FROM "CommonModel" WHERE company=$1 AND "modelId"=59`,[company]);return {...row,newGeneral:next.value}; }); const transactional=await Psql.tx(async tx=>{ await prepare(tx,'REPEATABLE READ'); const results={}; await tx.none('SAVEPOINT membership'); await tx.none('DELETE FROM "LessonRecord" WHERE company=$1 AND "objectId"=$2',[company,fixture.lesson]); results.deleted=await check(tx,company); const restored=await tx.one(`SELECT EXISTS(SELECT 1 FROM xs_payroll_sources_indexed($1) s WHERE s->>'lessonObjectId'=$2) value`,[company,fixture.fact]); if(!restored.value)throw Error('Fact did not reappear after canonical deletion'); await tx.none('ROLLBACK TO SAVEPOINT membership'); await tx.none('UPDATE "LessonRecord" SET company=$3 WHERE company=$1 AND "objectId"=$2',[company,fixture.lesson,'idx-tenant']); results.movedFrom=await check(tx,company);results.movedTo=await check(tx,'idx-tenant'); await tx.none('ROLLBACK TO SAVEPOINT membership');return results; }); const phantomId='ix'+Math.random().toString(36).slice(2,10); let announce,release; const firstReady=new Promise(resolve=>{announce=resolve;}),secondCommitted=new Promise(resolve=>{release=resolve;}); let firstError; const first=Psql.tx(async tx=>{ await prepare(tx); await tx.none('UPDATE "CommonModel" SET "generalId"=$3::numeric WHERE company=$1 AND "objectId"=$2',[company,fixture.common,fixture.newGeneral]); announce();await secondCommitted; }).catch(e=>{firstError=e;announce();throw e;});first.catch(()=>{}); try { await firstReady;if(firstError){release();await first;} let beforeCommit; try { await Psql.tx(async tx=>{ await prepare(tx); await tx.none(`INSERT INTO "LessonRecordFact" SELECT (jsonb_populate_record(NULL::"LessonRecordFact",to_jsonb(f)||jsonb_build_object('objectId',$3::text,'legacyGeneralId',$4::double precision))).* FROM "LessonRecordFact" f WHERE f.company=$1 AND f."objectId"=$2`,[company,fixture.fact,phantomId,fixture.newGeneral]); }); beforeCommit=await Psql.tx(async tx=>{await prepare(tx,'REPEATABLE READ');return check(tx,company);}); } finally {release();await first;} const afterCommit=await Psql.tx(async tx=>{ await prepare(tx,'REPEATABLE READ');const parity=await check(tx,company); const hidden=await tx.one(`SELECT EXISTS(SELECT 1 FROM "PayrollSourceIndex" WHERE company=$1 AND kind='fact' AND "objectId"=$2) indexed,EXISTS(SELECT 1 FROM xs_payroll_sources_indexed($1) s WHERE s->>'lessonObjectId'=$2) visible`,[company,phantomId]); if(!hidden.indexed||hidden.visible)throw Error('Concurrent phantom fact must remain indexed but hidden by current roots');return{...parity,...hidden}; }); return {transactional,phantom:{beforeCommit,afterCommit}}; } finally { release(); await Psql.tx(async tx=>{ await prepare(tx); await tx.none('DELETE FROM "LessonRecordFact" WHERE company=$1 AND "objectId"=$2',[company,phantomId]); await tx.none('UPDATE "CommonModel" SET "generalId"=$3::numeric WHERE company=$1 AND "objectId"=$2',[company,fixture.common,fixture.general]); }); } }