Skip to content

Hardening native rocksdb-js transaction log durability and retry-round drain tracking (follow-up to #3236) #3253

Closed

Description

@kriszyp

Follow-up to #3236.

Background

#3236 enforces in DatabaseTransaction that once a RocksDB transaction (or any link in a multi-store chain) has durably written its entry to the transaction log (hasDurableLog(transaction), tracked after a native transaction.commit() attempt with hasStagedLog), it is never aborted on conflict retries, commit-budget deadlines, record-lock lease expiry, or database-close commit suspension, and the caller stays awaiting the commit until it lands (matching sourceApply).

Native rocksdb-js changes and residual edges scoped out of #3236 into this follow-up issue:

  1. Native rocksdb-js conflict-before-log ordering in Transaction::Commit (e0c5e60a1aaf, 9dcabe0ba98a):

  2. Native rocksdb-js durable-log state exposure (committedPosition > 0) and post-log abort refusal/reconciliation (07a62c6fd60d, 20be4feb68b1, f933af77479c, 03d6ac4d1556ea87, 95ccdec124307a0d, 6e8fbfce53749284, 0533586f774f1afc):

    • Expose committedPosition > 0 (or a direct hasDurableLog accessor) from rocksdb-js on the native Transaction handle so DatabaseTransaction can read native log durability directly rather than inferring it from isRetry && hasStagedLog.
    • Harden Transaction::Abort() / Close() in rocksdb-js to refuse or safely reconcile handles that already hold a durable log entry (covering a first-attempt non-conflict commit failure after executeLogWork flushed the log entry, a synchronous throw from a custom onCommit hook during retry re-save, or a store close after WAIT_FOR_DRAIN_DEADLINE_MS).
  3. Tracking retry-round pre-commit completions awaits during waitForDrain() (e554d93b900e0d06):

    • Extend outstandingCommits tracking across the full retry round (including any asynchronous this.completions await added by a custom onCommit hook on retry).
  4. Propagating context.timestamp when creating a chained ImmediateTransaction link (fd98b84c9b2cb1a1, 8c1816872ad4f7f1):

    • Copy context.timestamp into timestamp and lastContextTimestamp when constructing a .next ImmediateTransaction link in Table.ts:txnForContext so the first write to a second database inherits the caller's version.
  5. Record-lock commit hold exclusion past commitConflictBudget (e7748380dccb1564):

    • retainForCommit defers local lock unlock only up to commitConflictBudget (limitation 2 in resources/record-locks.md); once Transaction::Commit validates conflicts before executeLogWork, conflicting commits no longer outlive commitConflictBudget.
  6. Direct raw eviction transaction commits in resources/Table.ts (ba9ebb8c973b5fe6, 9ea18b2db862):

    • Single-record and batched eviction in resources/Table.ts:4129-4139 and resources/Table.ts:9947-9960 commit raw RocksTransaction handles via commitTrackedRocksTransaction outside DatabaseTransaction and call transaction.abort() if commit() rejects with ERR_BUSY. When replicateEvictions is enabled (saveAudit: true), executeLogWork() flushes an 'evict' log entry before txn_->Commit() checks for conflict, so aborting or replacing the handle on ERR_BUSY orphans that 'evict' log entry until conflict-before-log ordering lands in rocksdb-js or raw eviction commits recommit the same handle.
  7. Per-save write-set rescan while a late immediate commit overlaps a retrying retained-iterator replay (6a76c39decccc6db):

    • In that overlap activeCommitHandle is unset and the replay handle differs from this.transaction, so every re-save in the round runs backfillActiveHandleWrites over the whole write set (O(N²) for that round). Ordinary commits and replay rounds backfill once per round; once conflict-before-log ordering lands, the retry rounds that reach this path mostly disappear.
  8. Edge cases of the stalled-commit monitor's logged-retry progress check (d5b79e28d4818b39, b6538f35ca1ef816, 3a5938b9fba8e158, 5151c6f2da9439a3):

    • The check spares a logged retry only while the summed retry counter of the chain's live logged handles keeps moving (activeDurableLogRetries() / deferForCommitInFlight() in resources/DatabaseTransaction.ts); a counter that stops for LOGGED_RETRY_STALL_MS (2 × the native park timeout) is treated as a stalled native commit. Diagnostics only, no data loss: an advancing or departing sibling can hide a wedged logged handle, and a handover to the next link's handle at an equal total can read as a stall.
    • The 2 × park window does not cover the up-to-1 s JS retry backoff once ROCKSDB_JS_PARK_TIMEOUT_MS is under about 1 s, and the parse of that variable differs from rocksdb-js's for non-integers and values under 50 ms.
    • abortNativeTransaction skips a handle flagged commitInFlight; a synchronous throw between the flag being set and the commit settling would leave it set (nothing on that path throws today).
    • Simplest direction if it matters: key the window on the per-handle retries of each live logged handle instead of one aggregate, and read the park timeout through rocksdb-js instead of mirroring its parser.

Review-Deferral: e0c5e60a1aaf severity=major category=reliability
Review-Deferral: 9dcabe0ba98a severity=major category=reliability
Review-Deferral: 07a62c6fd60d severity=nit category=code-quality
Review-Deferral: 20be4feb68b1 severity=nit category=code-quality
Review-Deferral: f933af77479c severity=minor category=test-coverage
Review-Deferral: 03d6ac4d1556ea87 severity=minor category=data-integrity
Review-Deferral: 95ccdec124307a0d severity=minor category=test-coverage
Review-Deferral: e554d93b900e0d06 severity=minor category=reliability
Review-Deferral: dd13e9a5c46676bd severity=nit category=test-coverage
Review-Deferral: 6e8fbfce53749284 severity=nit category=maintainability
Review-Deferral: 0533586f774f1afc severity=major category=reliability
Review-Deferral: fd98b84c9b2cb1a1 severity=nit category=correctness
Review-Deferral: 8c1816872ad4f7f1 severity=nit category=correctness
Review-Deferral: e7748380dccb1564 severity=nit category=concurrency
Review-Deferral: ba9ebb8c973b5fe6 severity=minor category=data-integrity
Review-Deferral: 6a76c39decccc6db severity=nit category=performance
Review-Deferral: d5b79e28d4818b39 severity=nit category=reliability
Review-Deferral: b6538f35ca1ef816 severity=nit category=reliability
Review-Deferral: 3a5938b9fba8e158 severity=nit category=reliability
Review-Deferral: 5151c6f2da9439a3 severity=nit category=reliability

🤖 Generated by Google Gemini, Claude Fable 5.1 and Claude Sonnet 5.5; posted via @kriszyp.

Activity

  1. added theissue type on Oct 19, 2026
  2. self-assigned this
    on Oct 24, 2026
  3. kriszyp commented on Oct 27, 2026

    @kriszyp
    MemberAuthor

    Confirmed on 2.12.0: a refused retained-handle write is replayed and replicated

    This is new real-binding evidence for this existing storage issue, discovered while testing the retry-conflict rule for HarperFast/harper-pro#980 (core #3527 / Pro HarperFast/harper-pro#1116). The slot-residency repair is stopped; no serialization or production retry-conflict guard was introduced. This issue already scopes the independent native/core fix, so I am adding the reproduction here instead of duplicating it.

    All three requested observations are positive: startup replay applies the abandoned PATCH to the primary and indexes; the real sender transmits it to a slot-0 owner; the real receiver commits it, and its row/index survive a flush and a fresh-process reopen.

    Observation Result
    Abort before the second native commit Same retained handle; synthetic retry-time conflict is thrown after re-save changes slot 0 to 1. Native abort throws ERR_TRANSACTION_ABANDONED; the independently committed PUT remains locally at region 2 / zone 0 / slot 0.
    Ordinary committed log query Returns the abandoned PATCH even without readUncommitted: true, after the concurrent PUT and abort. Its endTxn is true.
    Crash and startup replay SIGKILL after capturing the unflushed tail; a new process opens the root through getDatabases(), invoking resources/replayLogs.ts. Instrumented directCommitSync() shows the PATCH changing zone 0 to 1. Reconciliation of the following PUT leaves region 2 / zone 1 / slot 1; the zone index is 1.
    Actual sender replicateOverWS() resumes the origin log at the original PUT's key, sends a slot-0 reference and the abandoned {zone:1} PATCH, then the concurrent PUT and a barrier record. No bulk-copy fallback is used in the successful probe.
    Actual receiver A separate root is seeded with the original row via the internal copy-apply path. The captured sender buffers are passed unchanged to replicateOverWS() after handshake. The receiver writes the PATCH into the source-origin audit and changes the primary and zone index to 1; its final row in this probe is region 0 / zone 1 / slot 0. A drained flush and a third process confirm the same stored row/index.

    Reproduction details and limits

    Validated core 27ed145464184f6b7d80db293e18fb7702d06349, Pro 323bf0c042ee7834e4c51d00e002c777a4e98626, Node 26.2.0, actual resolved @harperfast/rocksdb-js 2.12.0 / RocksDB 11.8.1. Pro was rebuilt before probing. Native forceTryAgainForTesting(1) deterministically exercises the real binding's conflict path; this is not a mocked commit rejection.

    Table policy is region === 0 ? 0 : zone, with both region and zone indexed and no retained replica body. The initial PUT is {region:0,zone:0,body:'original'}. PATCH {zone:1} is admitted/audited for slot 0. After its first native commit reports TryAgain, an independent PUT writes {region:2,zone:0,body:'concurrent'} (still slot 0). The native retry now merges to slot 1, and a test-only wrapper around the staged write's commit throws a 409 before the retained handle can submit again. No production guard was added.

    The PATCH carries an explicit logical timestamp 60 seconds ahead, as in the existing retry-only reproduction. This ensures version reconciliation cannot simply supersede its effect with the concurrent PUT. The scope of this experiment is a post-log abort, not every successful retry or every application write. It demonstrates shared log/replay/replication exposure; it does not claim all TryAgain retries are abandoned.

    How the current readers distinguish abandonment

    They do not distinguish it in this binding:

    • The installed 2.12.0 TransactionLog.query() reads up to a log-wide _lastCommittedPosition, or physical file extent for readUncommitted. Its entry loop returns matching timestamp/data plus endTxn from the LAST flag. No commit/abandon discriminator is consulted for the returned batch. In this run the committed boundary includes the refused PATCH.
    • Core RocksTransactionLogStore.getRange() delegates to that native-package reader and handles corruption/exact-start boundaries, not abandoned writes.
    • replayLogs() asks for the unflushed tail with readUncommitted:true. A complete, decodable PATCH becomes _writeUpdate(..., false, ...) and is committed. endTxn proves the logged batch ended; it does not prove the original RocksDB transaction committed.
    • The replication sender iterates the ordinary log range and calls sendAuditRecord. The real captured wire result demonstrates there is no effective abandonment filter on this path.
    • Table version comparisons/reconciliation exist, but they are deduplication/order checks, not proof of commit. They did not suppress the refused PATCH in either observed apply path.

    The reader-first work in HarperFast/rocksdb-js#970 is relevant, but these probes used the installed unmodified 2.12.0 binding, not that PR. This report does not prescribe a new rollback contract or validate the pending fix.

    Captured evidence

    Primary/index state and transmitted records
    {
      "beforeCrash": {
        "slot": 0,
        "version": 1793106354330.2668,
        "flags": 234881600,
        "value": {
          "id": "retry-abort",
          "region": 2,
          "zone": 0,
          "body": "concurrent"
        },
        "indexes": {
          "region": [
            {
              "key": 2,
              "value": "retry-abort"
            }
          ],
          "zone": [
            {
              "key": 0,
              "value": "retry-abort"
            }
          ]
        }
      },
      "startupReplayCommits": [
        {
          "timestamp": 1793106354272.7883,
          "operations": [
            {
              "key": "retry-abort",
              "slot": 0,
              "skipped": false
            }
          ],
          "after": {
            "id": "retry-abort",
            "region": 0,
            "zone": 0,
            "body": "original"
          },
          "slot": 0
        },
        {
          "timestamp": 1793106354327.3306,
          "operations": [
            {
              "key": "retry-abort",
              "slot": 0,
              "skipped": false
            }
          ],
          "before": {
            "id": "retry-abort",
            "region": 0,
            "zone": 0,
            "body": "original"
          },
          "after": {
            "id": "retry-abort",
            "region": 0,
            "zone": 1,
            "body": "original"
          },
          "slot": 0
        },
        {
          "timestamp": 1793106354330.2668,
          "operations": [
            {
              "key": "retry-abort",
              "slot": 1,
              "skipped": false
            }
          ],
          "before": {
            "id": "retry-abort",
            "region": 0,
            "zone": 1,
            "body": "original"
          },
          "after": {
            "region": 2,
            "zone": 1
          },
          "slot": 1
        }
      ],
      "afterStartupReplay": {
        "slot": 1,
        "version": 1793106414325,
        "flags": 234947265,
        "value": {
          "region": 2,
          "zone": 1
        },
        "indexes": {
          "region": [
            {
              "key": 2,
              "value": "retry-abort"
            }
          ],
          "zone": [
            {
              "key": 1,
              "value": "retry-abort"
            }
          ]
        }
      },
      "sentToSlot0Owner": [
        {
          "type": "patch",
          "recordId": "retry-abort",
          "version": 1793106816709,
          "nodeId": 0,
          "value": {
            "zone": 1
          }
        },
        {
          "type": "put",
          "recordId": "retry-abort",
          "version": 1793106756712.71,
          "nodeId": 0,
          "value": {
            "id": "retry-abort",
            "region": 2,
            "zone": 0,
            "body": "concurrent"
          }
        },
        {
          "type": "put",
          "recordId": "barrier",
          "version": 1793106756720.4285,
          "nodeId": 0,
          "value": {
            "id": "barrier",
            "region": 0,
            "zone": 0,
            "body": "after abort"
          }
        }
      ],
      "slotReferences": [
        [
          130,
          {
            "slot": 0,
            "mapVersion": 1,
            "mapDigest": "83efbf3e0367b5f5509b37526eb1fc00278adb73ef001933aad8791c284d4b45"
          },
          2147483648
        ]
      ],
      "receiverAfterFlushAndReopen": {
        "primary": {
          "slot": 0,
          "version": 1793106816709,
          "flags": 234881728,
          "value": {
            "id": "retry-abort",
            "region": 0,
            "zone": 1,
            "body": "original"
          },
          "indexes": {
            "region": [
              {
                "key": 0,
                "value": "barrier"
              },
              {
                "key": 0,
                "value": "retry-abort"
              }
            ],
            "zone": [
              {
                "key": 0,
                "value": "barrier"
              },
              {
                "key": 1,
                "value": "retry-abort"
              }
            ]
          }
        },
        "audit": [
          {
            "type": "patch",
            "slot": 0,
            "version": 1793106816709,
            "txnLogKey": 1793106756710.8628,
            "endTxn": true,
            "value": {
              "zone": 1
            }
          }
        ]
      }
    }

    Reproducible probe scripts

    Place these three files together in any scratch directory inside the Pro worktree, and run from the Pro root with its core companion checked out. They use only fresh roots passed on the command line. Build core and Pro first. The write process intentionally dies by SIGKILL; that is the expected result, not a test-runner failure. A successful probe exit records observations, not a passing assertion that abort is safe.

    node <scratch>/round15-replay.cjs write <fresh-replay-root> <before.json>
    node <scratch>/round15-replay.cjs replay <same-replay-root> <after.json>
    node <scratch>/round15-replication.cjs send <fresh-sender-root> <sent.json>
    node <scratch>/round15-replication.cjs receive <fresh-receiver-root> <received.json> <sent.json>
    node <scratch>/round15-replication.cjs verify <same-receiver-root> <reopened.json>
    round15-abort.cjs
    const assert = require('node:assert');
    exports.snapshot = (Records, id = 'retry-abort') => {
     const row = Records.primaryStore.getEntry(id);
     return { slot: row?.residencySlot, version: row?.version, flags: row?.metadataFlags, value: row?.value,
      indexes: Object.fromEntries(['region', 'zone'].map(name => [name, [...Records.indices[name].getRange({})]])) };
    };
    exports.audit = (Records, options = {}) => [...Records.auditStore.getRange({start:1, log:0, ...options})]
     .filter(e => e.tableId === Records.tableId && e.recordId === 'retry-abort')
     .map(e => ({type:e.type, slot:e.residencySlot, version:e.version, txnLogKey:e.txnLogKey, endTxn:e.endTxn, value:e.getValue(Records.primaryStore)}));
    exports.abortMove = async (Records, api) => {
     const {transaction, Transaction, constants, setResidencyPlacementGuard, replicationConfirmation} = api;
     const binding = Object.values(require.cache).find(m => m.filename?.endsWith('.node') && typeof m.exports.forceTryAgainForTesting === 'function')?.exports;
     assert.ok(binding);
     const id = 'retry-abort';
     await Records.put(id, {id,region:0,zone:0,body:'original'});
     const nativeCommit = Transaction.prototype.commit, nativeAbort = Transaction.prototype.abort;
     const attempts = [], aborts = [], confirmations = [];
     let originalId, admittedSlot, rejectedSlot;
     const context = {timestamp:Date.now()+60_000};
     const conflict = Object.assign(new Error(`Slot changed during native retry for table ${Records.databaseName}.${Records.tableName} record ${id}; retry the write`),{statusCode:409,retryable:true});
     Transaction.prototype.abort = function(...args) {
      const id = this.id;
      try { return nativeAbort.apply(this,args); }
      catch(e) { aborts.push({id,code:e.code,message:e.message}); throw e; }
     };
     Transaction.prototype.commit = function(...args) {
      if(this.store?.db !== Records.primaryStore.store.db) return nativeCommit.apply(this,args);
      attempts.push(this.id);
      const first = originalId === undefined;
      if(first) originalId = this.id;
      const result = nativeCommit.apply(this,args);
      if(!first) return result;
      const changeBase = () => transaction({}, () => Records.put(id,{id,region:2,zone:0,body:'concurrent'}));
      return result.then(async outcome => { assert.equal(outcome,constants.RETRY_NOW_VALUE); await changeBase(); return outcome; },
       async error => { assert.equal(error.code,'ERR_TRY_AGAIN'); await changeBase(); throw error; });
     };
     setResidencyPlacementGuard(() => {});
     replicationConfirmation((...args) => { confirmations.push(args); return Promise.resolve(); });
     try {
      await assert.rejects(transaction(context,async () => {
       await Records.patch(id,{zone:1});
       const write = context.transaction.writes.find(w => w?.key === id);
       admittedSlot = write.residencyOwnerSlot;
       assert.equal(admittedSlot,0);
       const save = write.commit;
       write.commit = function(...args) {
        const result = save.apply(this,args);
        if(args[3]?.isRetry && write.stagedEntry.residencySlot !== admittedSlot) {
         rejectedSlot = write.stagedEntry.residencySlot; throw conflict;
        }
        return result;
       };
       binding.forceTryAgainForTesting(1);
      }), error => error === conflict);
      assert.equal(rejectedSlot,1);
      assert.equal(attempts.filter(id => id === originalId).length,1);
      assert.equal(Records.primaryStore.getEntry(id).value.zone,0);
      return {originalId,admittedSlot,rejectedSlot,attempts,aborts,confirmations,
       primary:exports.snapshot(Records), committed:exports.audit(Records),uncommitted:exports.audit(Records,{readUncommitted:true})};
     } finally {
      binding.forceTryAgainForTesting(0); Transaction.prototype.commit = nativeCommit; Transaction.prototype.abort = nativeAbort;
      setResidencyPlacementGuard(undefined); replicationConfirmation(undefined);
     }
    };
    round15-replay.cjs
    const {createRequire}=require('node:module');
    const path=require('node:path');
    const {mkdirSync,writeFileSync}=require('node:fs');
    const assert=require('node:assert');
    const req=createRequire(path.resolve('core/package.json'));
    const [mode,root,output]=process.argv.slice(2);
    const env=req('#src/utility/environment/environmentManager');
    const terms=req('#src/utility/hdbTerms');
    env.setProperty(terms.HDB_SETTINGS_NAMES.HDB_ROOT_KEY,root);
    env.setProperty(terms.CONFIG_PARAMS.STORAGE_PATH,path.join(root,'database'));
    env.setProperty('replication_shards',[0,1]);env.setProperty('replication_shard',0);
    env.setProperty(terms.CONFIG_PARAMS.LOGGING_AUDITLOG,true);
    mkdirSync(path.join(root,'database'),{recursive:true});
    const {DatabaseTransaction,replicationConfirmation}=req('#src/resources/DatabaseTransaction');
    const helper=require('./round15-abort.cjs');
    const commits=[];
    if(mode==='replay') {
     const original=DatabaseTransaction.prototype.directCommitSync;
     DatabaseTransaction.prototype.directCommitSync=function(...args) {
      if(!this.isReplay) return original.apply(this,args);
      const before=this.db.getEntry('retry-abort');
      const operations=this.writes.filter(Boolean).map(w=>({key:w.key,slot:w.stagedEntry?.residencySlot,skipped:w.skipped}));
      const result=original.apply(this,args);
      const after=this.db.getEntry('retry-abort');
      commits.push({timestamp:this.timestamp,operations,before:before?.value,after:after?.value,slot:after?.residencySlot});
      return result;
     };
    }
    const {getDatabases}=req('#src/resources/databases');
    const {loadGQLSchema}=req('#src/resources/graphql');
    const {getResidencyMapTable,setResidencyPlacementGuard}=req('#src/resources/residencyMaps');
    const {server}=req('#src/server/Server');
    req('#src/server/threads/manageThreads').setMainIsWorker(true);
    server.shards=new Map([[0,[{name:server.hostname}]],[1,[{name:'other'}]]]);
    (async()=>{
     if(mode==='write') {
      await loadGQLSchema('type RetryRows @table(database: "slotabort", replicaExpiration: 0) @sharded(slots: 2, by: "region", slotFrom: "region === 0 ? 0 : zone") { id: ID @primaryKey region: Int @indexed zone: Int @indexed body: String }');
      const Records=getDatabases().slotabort.RetryRows;
      await getResidencyMapTable().primaryStore.rootStore.flush();
      await Records.primaryStore.rootStore.flush();
      const result=await helper.abortMove(Records,{
       transaction:req('#src/resources/transaction').transaction,
       ...req('./unitTests/rocksdbBindings.mts').rocksdbBindings(),setResidencyPlacementGuard,replicationConfirmation
      });
      result.replayTail=helper.audit(Records,{start:undefined,startFromLastFlushed:true,readUncommitted:true});
      assert.ok(result.replayTail.some(e=>e.type==='patch'));
      writeFileSync(output,JSON.stringify(result,null,2));
      process.kill(process.pid,'SIGKILL');
     } else {
      const Records=getDatabases().slotabort.RetryRows;
      const result={commits,primary:helper.snapshot(Records),audit:helper.audit(Records,{readUncommitted:true})};
      assert.ok(commits.some(c=>c.operations.some(w=>w.key==='retry-abort')),'startup replay reached the target record');
      writeFileSync(output,JSON.stringify(result,null,2));
      process.exit(0);
     }
    })().catch(e=>{console.error(e);process.exit(1);});
    round15-replication.cjs
    const {createRequire}=require('node:module');
    const path=require('node:path');
    const {mkdirSync,writeFileSync,readFileSync}=require('node:fs');
    const {EventEmitter}=require('node:events');
    const assert=require('node:assert');
    const req=createRequire(path.resolve('package.json'));
    const [mode,root,output,sourceFile]=process.argv.slice(2);
    const env=req('#src/core/utility/environment/environmentManager');
    const terms=req('#src/core/utility/hdbTerms');
    env.setProperty(terms.HDB_SETTINGS_NAMES.HDB_ROOT_KEY,root);
    env.setProperty(terms.CONFIG_PARAMS.STORAGE_PATH,path.join(root,'database'));
    env.setProperty('replication_shards',[0,1]);env.setProperty('replication_shard',0);
    env.setProperty(terms.CONFIG_PARAMS.LOGGING_AUDITLOG,true);
    env.setProperty(terms.CONFIG_PARAMS.NODE_HOSTNAME,mode==='send'?'abort-source':'slot0-peer');
    mkdirSync(path.join(root,'database'),{recursive:true});
    const {server}=req('#src/core/server/Server');
    req('#src/core/server/threads/manageThreads').setMainIsWorker(true);
    const {transaction}=req('#src/core/resources/transaction');
    const {replicationConfirmation,DatabaseTransaction,getOutstandingCommits}=req('#src/core/resources/DatabaseTransaction');
    const {table}=req('#src/core/resources/databases');
    const {initializeResidencySlots,setResidencyPlacementGuard,getResidencyConfiguration}=req('#src/core/resources/residencyMaps');
    const {setReplicator}=req('#src/replication/replicator');
    const {databaseSubscriptions,replicateOverWS}=req('#src/replication/replicationConnection');
    const {buildLocalCapabilities}=req('#src/replication/protocolCapabilities');
    const {advanceShardingHistory}=req('#src/replication/shardingBootstrapState');
    const {shardingLayout}=req('#src/replication/shardingRecovery');
    const {encode,decode}=req('msgpackr');
    const {Decoder,readAuditEntry}=req('#src/core/resources/auditStore');
    const helper=require('./round15-abort.cjs');
    class Socket extends EventEmitter {
     readyState=1;sent=[];closes=[];_socket={bytesRead:0,bytesWritten:0,setMaxListeners(){}};
     send(data){this.sent.push(Buffer.from(data));}
     close(code=1000,reason=''){if(this.readyState===3)return;this.readyState=3;this.closes.push({code,reason});this.emit('close',code,Buffer.from(reason));}
     terminate(){this.close(1006);} ping(){} pause(){} resume(){}
    }
    async function until(predicate){const end=Date.now()+10000;while(!predicate()){assert.ok(Date.now()<end,'replication timeout');await new Promise(r=>setTimeout(r,10));}}
    function records(socket,Records){
     const entries=[];
     for(const frame of socket.sent){
      if(frame[0]>127)continue;
      const decoder=new Decoder(frame.buffer,frame.byteOffset,frame.byteLength);decoder.position=8;
      while(decoder.position<frame.length){const length=decoder.readInt(),start=decoder.position;decoder.position+=length;if(length===9&&frame[start]===11)continue;
       const e=readAuditEntry(frame,start,start+length);
       if(e.recordId==='retry-abort'||e.recordId==='barrier')entries.push({type:e.type,recordId:e.recordId,slot:e.residencySlot,version:e.version,nodeId:e.nodeId,value:e.getValue(Records.primaryStore)});
      }
     }
     return entries;
    }
    (async()=>{
    
    server.shards=new Map([[0,[{name:'abort-source'},{name:'slot0-peer'}]],[1,[{name:'slot1-owner'}]]]);
    server.nodes=[{name:mode==='send'?'slot0-peer':'abort-source',replicates:true}];
    
     const database='slotabort';
     await initializeResidencySlots({database,slots:2,ranges:[{start:0,end:1,shard:0},{start:1,end:2,shard:1}]});
     const Records=table({database,table:'RetryRows',audit:true,replicaExpiration:0,residency:{slots:2,by:'region',byAttribute:true,slotFrom:'region === 0 ? 0 : zone'},attributes:[{name:'id',isPrimaryKey:true},{name:'region',indexed:true},{name:'zone',indexed:true},{name:'body'}]});
     for(const shard of [0,1]) advanceShardingHistory(Records.dbisDB,shardingLayout(database,shard),{initializer:server.hostname,agreement:'fixture',run:'fixture',phase:'began',localFloor:1,localOrigin:server.hostname});
     setReplicator(database,Records,{});
     await until(()=>databaseSubscriptions.get(database)?.auditStore);
     if(mode==='verify'){writeFileSync(output,JSON.stringify({primary:helper.snapshot(Records),audit:helper.audit(Records,{log:undefined,readUncommitted:true})},null,2));process.exit(0);}
     const socket=new Socket();
     const peer=mode==='send'?'slot0-peer':'abort-source';
     replicateOverWS(socket,{}, {name:peer,replicates:true});
     socket.emit('message',encode([140,peer,database,[],buildLocalCapabilities(90000,true,true),{shardsDigest:getResidencyConfiguration().digest}]));
     await until(()=>socket.sent.some(f=>f[0]>127&&decode(f)[0]===140)||socket.closes.length);
     assert.deepEqual(socket.closes,[]);
     console.log('CONFIG',server.hostname,getResidencyConfiguration(),[...server.shards]);
     if(mode==='send'){
      const api={transaction,replicationConfirmation,setResidencyPlacementGuard};
      // Resolve the exact binding used by the compiled Pro core.
      const native=req('@harperfast/rocksdb-js');api.Transaction=native.Transaction;api.constants=native.constants;
      const result=await helper.abortMove(Records,api);
      socket.emit('message',encode([129,[{name:server.hostname,startTime:result.committed[0].txnLogKey,tables:[],replicateByDefault:true}],[]]));
      await Records.put('barrier',{id:'barrier',region:0,zone:0,body:'after abort'});
      await until(()=>socket.closes.length||records(socket,Records).some(e=>e.recordId==='barrier'));
      result.closes=socket.closes;result.sent=records(socket,Records);result.frames=socket.sent.map(f=>f.toString('base64'));
      result.messages=socket.sent.filter(f=>f[0]>127).map(f=>decode(f));
      writeFileSync(output,JSON.stringify(result,null,2));
      assert.deepEqual(socket.closes,[]);
     }else{
      const source=JSON.parse(readFileSync(sourceFile,'utf8'));
      // Observe completed received commits via the table's real primary after each inbound frame drains.
      const seed=source.committed[0];
      const nodeId=req('#src/core/resources/nodeIdMapping').getIdOfRemoteNode('abort-source',Records.auditStore);
      const context={source:{},sourceApply:true,timestamp:seed.txnLogKey};
      await transaction(context,async()=>{
       const row=await Records.getResource('retry-abort',context);
       row._writeUpdate('retry-abort',seed.value,true,{nodeId,version:seed.version,isCopyApply:true,residencySlot:0});
       row.save();
      });
      const frames=source.frames.map(f=>Buffer.from(f,'base64'));
      for(const frame of frames){
       if(frame[0]>127 && [140,129,142].includes(decode(frame)[0]))continue;
       socket.emit('message',frame);
      }
      await until(()=>socket.closes.length||Records.primaryStore.getEntry('barrier'));
      await until(()=>getOutstandingCommits().count===0);
      await req('#src/core/resources/databases').flushDatabases();
      const result={closes:socket.closes,primary:helper.snapshot(Records),audit:helper.audit(Records,{log:undefined,readUncommitted:true}),responses:socket.sent.filter(f=>f[0]>127).map(f=>decode(f))};
      writeFileSync(output,JSON.stringify(result,null,2));
      assert.deepEqual(socket.closes,[]);
     }
     socket.close();process.exit(0);
    })().catch(e=>{console.error(e);process.exit(1);});

    🤖 Generated by OpenAI Codex; posted via @kriszyp.

  4. kriszyp commented on Nov 3, 2026

    @kriszyp
    MemberAuthor

    Closing as completed. The native hardening this issue tracked shipped in rocksdb-js 2.13.0 and is adopted on main for 5.4.

    rocksdb-js 2.13.0

    Harper

    Known gap (accepted): retry-dedup certification is kept in memory, per process and per log. After a restart, dedup on a log stays off until that log's pre-restart files are purged, about one audit retention (3 days by default). During that window an increment or CRDT add can be double-counted, the same as before #3928.

    Follow-ups

    🤖 Generated by Claude Code (Claude Opus 5.5); posted via @kriszyp.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

No labels
No labels

Type

Fields

Priority

P2

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions