xiaoshu-app/scripts/tests/payroll-source-index-concurrency.mjs

// 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]);
    });
  }
}