feat(replication): add replicationState.awaitDocumentPushed()

[pubkey]

Summary

Implements the feature requested in #8632: a per-document push acknowledgement on the replication state.

replicationState.awaitDocumentPushed(rxDocument) returns a promise that resolves once the given RxDocument instance was successfully pushed to the server.

const doc = await myCollection.insert({ id: 'foobar', value: 10 });
await myReplicationState.awaitDocumentPushed(doc);
// here we know that this document state reached the master

How it works

Each RxDocument carries its write time in _meta.lwt. The upstream (push) checkpoint has the default RxStorage checkpoint shape { id, lwt } and moves forward with every successful push, in the sort order [_meta.lwt ASC, primaryKey ASC]. A document state is considered pushed once it is no longer "after" the checkpoint, which mirrors the comparison used by getChangedDocumentsSinceQuery().

The implementation:

• does an immediate check against the current upstream checkpoint
• otherwise re-checks on each sent$ emission (when a document is sent) and when the replication becomes idle (active$ → false), at which point the checkpoint for that push cycle has been written
• if the document was overwritten by a newer local write before it could be pushed, the promise still resolves once any later state (or any document with a higher write time) was pushed, since that proves the given state reached the server

No timeout is built in by design. The docs show how to combine it with Promise.race().

Behavior notes

• Calling it on a pull-only replication throws RC_PUSH_AWAIT.
• If the replication is canceled before the document was pushed, the promise stays pending, matching the behavior of awaitInitialReplication().

Changes

• src/plugins/replication/index.ts: new awaitDocumentPushed() method and isDocumentStateOlderThenCheckpoint() helper
• src/plugins/dev-mode/error-messages.ts: new error code RC_PUSH_AWAIT
• test/unit/replication.test.ts: tests for resolve-after-push, resolve-immediately-if-already-pushed, not-resolve-before-push, document-inserted-before-replication, and the pull-only throw
• docs-src/docs/replication.md: new awaitDocumentPushed() section with a Promise.race() timeout example
• changelog entry

Closes #8632

🤖 Generated with Claude Code

---

Generated by Claude Code

[pubkey]

@copilot inline this to make the example shorter

[Copilot]

Addressed in 9059aaf.

[github-actions]

✅ Verify Test Reproduction: Tests FAILED without the fix (expected)

This confirms the changed tests correctly reproduce the bug that the source changes fix.

This workflow runs the changed tests without the source fix to verify they reproduce the bug.

Show output

...(truncated, showing last 200 of 3075 lines)
      �[32m✓ �[39mshould clean up the deleted documents
      �[32m✓ �[39mshould work by manually calling RxCollection.cleanup()
    cleanup and replication
      �[32m✓ �[39mshould pause the cleanup when a replication is not in sync
      �[32m✓ �[39mshould also run a cleanup on the replication state meta data
    issues
      �[32m✓ �[39mminimumDeletedTime not respected
      �[32m✓ �[39mshould correctly loop cleanup when storage cleanup returns false (batched cleanup)
      �[32m✓ �[39mcleanup() should return true as stated by the TypeScript return type
      �[32m✓ �[39mfields with umlauts and emojis could break the state after cleanup in some storages

  hooks.test.js
    get/set
      �[32m✓ �[39mshould set a hook
      �[32m✓ �[39mshould get a hook
      �[32m✓ �[39mshould get a parallel hook
    insert
      pre
        positive
          �[32m✓ �[39mseries
          �[32m✓ �[39mparallel
          �[32m✓ �[39mshould save a modified document
          �[32m✓ �[39masync: should save a modified document
          �[32m✓ �[39mshould not insert if hook throws
          �[32m✓ �[39mshould have the collection bound to the this-scope
      post
        positive
          �[32m✓ �[39mseries
          �[32m✓ �[39mparallel
          �[32m✓ �[39mshould call post insert hook after bulkInsert
    save
      pre
        positive
          �[32m✓ �[39mseries
          �[32m✓ �[39mparallel
          �[32m✓ �[39mshould save a modified document
          �[32m✓ �[39masync: should save a modified document
          �[32m✓ �[39mshould not save if hook throws
      post
        positive
          �[32m✓ �[39mseries
          �[32m✓ �[39mparallel
          �[32m✓ �[39mshould receive the RxDocument instance as second argument
    remove
      pre
        positive
          �[32m✓ �[39mseries
          �[32m✓ �[39mparallel
          �[32m✓ �[39mshould not remove if hook throws
          �[32m✓ �[39mshould call pre remove hook before bulkRemove
          �[32m✓ �[39mshould keep the field value that was added by the hook
      post
        positive
          �[32m✓ �[39mseries
          �[32m✓ �[39mparallel
          �[32m✓ �[39mshould have the collection bound to the this-scope
          �[32m✓ �[39mshould call post remove hook after bulkRemove
    postCreate
      positive
        �[32m✓ �[39mshould define a getter
      negative
        �[32m✓ �[39mshould throw when adding an async-hook
    issues
      �[32m✓ �[39mISSUE #158 : Throwing error in async preInsert does not prevent insert
      �[32m✓ �[39mshould auto-generate composite primary key when set in preInsert hook

  rx-pipeline.test.js
    basics
      �[32m✓ �[39madd and remove a pipeline
      �[32m✓ �[39mwrite some document depending on another
      �[32m✓ �[39mwrite some document depending on another with schema validator
      �[32m✓ �[39mshould store the transformed data to the destination
    .awaitIdle()
      �[32m✓ �[39mshould have updated its internal timestamps
    error handling
      �[32m✓ �[39mshould not swallow the error if the handler throws
      �[32m✓ �[39mshould not swallow the error if the handler throws (async)
      �[32m✓ �[39mshould not break reads on destination after handler throws
    checkpoints
      �[32m✓ �[39mshould continue from the correct checkpoint
    multiInstance
      �[32m✓ �[39mshould only run the pipeline at the leader
      �[32m✓ �[39mshould halt reads on other tab while pipeline is running
    transactional behavior
      �[32m✓ �[39mshould not block reads/writes that come from inside the pipeline handler
      �[32m✓ �[39mshould not block reads that come from inside the pipeline handler when already cached outside
      �[32m✓ �[39mshould be able to do writes dependent on reads
      �[32m✓ �[39mshould not block reads when localDocument inserted
    multiple pipelines to same destination
      �[32m✓ �[39mshould not deadlock when two pipelines write to the same destination and both handlers read from it
    .remove()
      �[32m✓ �[39mshould properly clean up checkpoint when remove() is called while pipeline is processing

  orm.test.js
    statics
      create
        positive
          �[32m✓ �[39mcreate a collection with static-methods
        negative
          �[32m✓ �[39mcrash when name not allowed (startsWith(_))
          �[32m✓ �[39mcrash when name not allowed (name reserved)
      run
        �[32m✓ �[39mshould be able to run the method
        �[32m✓ �[39mshould have the right this-context
        �[32m✓ �[39mshould be able to use this.insert()
    instance-methods
      create
        positive
          �[32m✓ �[39mcreate a collection with instance-methods
          �[32m✓ �[39mthis-scope should be bound to document
        negative
          �[32m✓ �[39mcrash when name not allowed (startsWith(_))
          �[32m✓ �[39mcrash when name not allowed (name reserved)
          �[32m✓ �[39mcrash when name not allowed (name is top-level field in schema)
      run
        �[32m✓ �[39mshould be able to run the method
        �[32m✓ �[39mshould have the right this-context
        �[32m✓ �[39mshould not be confused with many collections
    ISSUES
      �[32m✓ �[39mattachment ORM method with built-in name should throw
      �[32m✓ �[39mORM method with populate-getter suffix should throw COL18
      �[32m✓ �[39m#791 Document methods are not bind() to the document

  replication-protocol.test.ts (implementation: dexie)
    helpers
      checkpoint
        �[32m✓ �[39m#3627 should not write a duplicate checkpoint
    down
      �[32m✓ �[39mit should write the initial data and also the ongoing insert
    up
      �[32m✓ �[39mit should write the initial data and also the ongoing insert
      �[32m✓ �[39mshould replicate the insert and the update and the delete
    different configurations
      �[32m✓ �[39mshould be able to replicate A->Master<-B
      �[32m✓ �[39mshould be able to replicate A->B->C->Master
      �[32m✓ �[39mon multi instance it should be able to mount on top of the same storage config with a different instance
      �[32m✓ �[39mshould respect the given local start checkpoint
    conflict handling
      �[32m✓ �[39mboth have inserted the exact same document -> no conflict handler must be called
      �[32m✓ �[39mboth have inserted the same document with different properties
      �[32m✓ �[39mboth have updated the document with different values
      �[32m✓ �[39mdoing many writes on the fork should not lead to many writes on the master
    attachment replication
      �[32m✓ �[39mpush-only: should replicate the attachments to master
      �[32m✓ �[39mpull-only: should replicate the attachments to fork
    stability
      �[32m✓ �[39mBUG: writes to both sides can make it not end up with the correct state
      �[32m✓ �[39mdo many writes while replication is running (0)
      �[32m✓ �[39mdo many writes while replication is running (1)
      �[32m✓ �[39mdo many writes while replication is running (2)
      �[32m✓ �[39mdo many writes while replication is running (3)
      �[32m✓ �[39mdo many writes while replication is running (4)
    issues
      �[32m✓ �[39mshould be able to replicate local documents
      �[32m✓ �[39mshould not stuck when replicating many document in the initial replication

  replication.test.ts
    init
      �[32m✓ �[39mcreate storage
    non-live replication
      �[32m✓ �[39mshould replicate both sides
      �[32m✓ �[39mshould allow asynchronous push and pull modifiers
      �[32m✓ �[39mshould skip the document when the push-modifier returns null
      �[32m✓ �[39msent$ must not emit null when the push-modifier returns null
      �[32m✓ �[39msent$ must emit documents in WithDeleted format with _deleted field when deletedField is custom
      �[32m✓ �[39mshould not save pulled documents that do not match the schema
      �[32m✓ �[39mshould not update the push checkpoint when the conflict handler returns invalid documents that do not match the schema
      �[32m✓ �[39mshould never resolve awaitInitialReplication() on erroring replication
      �[32m✓ �[39mshould never resolve awaitInitialReplication() on canceled replication
    live replication
      �[32m✓ �[39mshould replicate all writes
      �[32m✓ �[39m#3994 should respect the push.batchSize
      �[32m✓ �[39mshould emit active$ when a replication cycle is running
    other
      �[32m✓ �[39mshould clean up the replication meta storage when the get collection gets removed
      �[32m✓ �[39mshould respect the initial push checkpoint
      �[32m✓ �[39mshould respect the initial pull checkpoint
      autoStart
        �[32m✓ �[39mshould run first replication by default
        �[32m✓ �[39mshould not run first replication when autoStart is set to false
      .awaitInSync()
        �[32m✓ �[39mshould resolve after some time
        �[32m✓ �[39mshould never resolve when offline
      .awaitDocumentPushed()
        �[31m✗ �[39m�[31mshould resolve after the document was pushed to the master�[39m
	TypeError: replicationState.awaitDocumentPushed is not a function
	    at Context.eval (webpack://rxdb/./test_tmp/unit/replication.test.js?:756:32)


Chrome Headless 149.0.0.0 (Linux 0.0.0): Executed 983 of 1327�[31m (1 FAILED)�[39m (33.088 secs / 32.65 secs)
�[31mTOTAL: 1 FAILED, 982 SUCCESS�[39m


�[31m1) should resolve after the document was pushed to the master
�[39m�[31m     replication.test.ts other .awaitDocumentPushed()
�[39m�[90m     TypeError: replicationState.awaitDocumentPushed is not a function
    at Context.eval (webpack://rxdb/./test_tmp/unit/replication.test.js?:756:32)
�[39m

View full workflow run

[pubkey]

explain that another older or newer RxDocument instance of that document is not the exact same RxDcoument. a RxDocument represents the state of the document in a given point in time @copilot

[Copilot]

Added in 1eef39d.

[pubkey]

@copilot here we do not need the hint about timeout

[Copilot]

Removed in ea616a9.

[pubkey]

@copilot dont we have myRxDoucment.primary?

[Copilot]

Yes, updated to use doc.primary in ea616a9.

[pubkey]

@copilot shouldnt we use myRxReplicationState.sent$ here?

[pubkey]

Done in 03ff70e, now triggered by sent$.

One note: sent$ is used as the trigger, but I kept the isDocumentStateOlderThenCheckpoint() comparison as the actual resolve condition instead of resolving directly on the emitted document. Two reasons:

• sent$ emits WithDeleted<RxDocType> without _meta, so it does not carry the _meta.lwt needed to match the exact instance version.
• sent$ fires before the push checkpoint is persisted, and matching only by primary key could resolve on an older in-flight push of the same document. Comparing against the checkpoint (after await streamQueue.up) keeps the point-in-time precision.

The same commit also renames the helper to isDocumentStateOlderThenCheckpoint and adds the test where the insert happens before the replication state is created.

---

Generated by Claude Code

[Copilot]

Changed to use sent$ and active$ in 03ff70e.

[pubkey]

@copilot call thsi function isDocumentStateOlderThenCheckpoint

[Copilot]

Renamed in 03ff70e.

[pubkey]

@copilot add test where the insert happens before the replication state is created

[Copilot]

Added in 03ff70e.

[pubkey]

we do not need that hint here in the docs

[pubkey]

i do not think we should await this here, before isPushed is called another cycled could have started already that does not affect our doucment anyways.

[pubkey]

Removed in b5dc0f0. You are right.

I dropped the streamQueue.up await entirely and now resolve directly on the sent$ emission whose primary key matches the document. That emission already proves the document reached the master, so there is no need to wait for any cycle or re-read the checkpoint. The isDocumentStateOlderThenCheckpoint() comparison is now only used for the initial check (documents that were already pushed before awaitDocumentPushed() was called).

---

Generated by Claude Code