Preserve committed entities across batches without rewriting history

[DZakh]

Summary

This change ensures that entities committed in previous batches are preserved in the in-memory store and can be read in subsequent batches without hitting the database, while maintaining correct history tracking.

Key Changes

• InMemoryStore: Added commitedCheckpointId field to track the last committed checkpoint, replacing the previous approach of checking against loadedFromDbCheckpointId
• Batch write logic: Modified to selectively clear in-memory state while preserving committed entities:
  • Clears raw events, effects, and rollback state after each batch
  • Keeps latestEntityChangeById for entities below a size threshold (50,000 entities)
  • Updates commitedCheckpointId to the last checkpoint in the batch
• InMemoryTable.Entity:
  • Added resetButKeepLatestChanges() to drop per-batch state while preserving committed entities
  • Updated set() to accept commitedCheckpointId parameter for accurate change tracking
• Change detection: Updated logic to compare checkpoint IDs against commitedCheckpointId instead of loadedFromDbCheckpointId, ensuring only uncommitted changes are written to the database
• Test coverage: Added comprehensive test verifying that entities from previous batches are read from memory and history is correctly maintained across batches

Implementation Details

• When a batch completes, the store now keeps the latest entity changes in memory if the table size is below the limit, allowing subsequent batches to read them without database queries
• The commitedCheckpointId acts as a watermark to distinguish between committed changes (which should be preserved) and in-batch changes (which should be written)
• For large entity tables (≥50,000 entities), the store still clears the in-memory table to prevent unbounded memory growth on long-running indexers

https://claude.ai/code/session_01JRHFPTtYobx3WjahffvE1y

Summary by CodeRabbit

• New Features

  • Improved in-memory checkpoint tracking so committed entity state is preserved across batch writes while retaining recent change history.

• Bug Fixes

  • Rollback and entity update flows refined to align in-memory state with committed checkpoints, preventing unintended history truncation.

• Tests

  • Added a test validating entity persistence and history ordering across sequential batches.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: da17971b-2057-4a64-ab4a-118f789fd237

📥 Commits

Reviewing files that changed from the base of the PR and between 0e6a951 and c3179d1.

📒 Files selected for processing (4)

• packages/envio/src/InMemoryStore.res
• packages/envio/src/InMemoryTable.res
• packages/envio/src/UserContext.res
• scenarios/test_codegen/test/helpers/MockIndexer.res

🚧 Files skipped from review as they are similar to previous changes (4)

• scenarios/test_codegen/test/helpers/MockIndexer.res
• packages/envio/src/UserContext.res
• packages/envio/src/InMemoryTable.res
• packages/envio/src/InMemoryStore.res

---

📝 Walkthrough

Walkthrough

Store now tracks committedCheckpointId (initialized from Internal.initialCheckpointId), uses it to filter and retain entity changes across batch writes, adds a table helper to keep latest changes, and threads the committed checkpoint through rollback and in-memory write paths.

Changes

Incremental Entity Update Tracking

Layer / File(s)	Summary
Store state and initialization with committed checkpoint packages/envio/src/InMemoryStore.res	InMemoryStore.t adds mutable committedCheckpointId: Internal.checkpointId, initialized from Internal.initialCheckpointId; keepLatestChangesLimit constant added for post-write trimming.
Entity table checkpoint awareness and reset helper packages/envio/src/InMemoryTable.res	Adds Entity.resetButKeepLatestChanges and changes Entity.set signature to accept ~committedCheckpointId, updating logic that accumulates prevEntityChanges to use a range comparison against the committed checkpoint.
Batch write incremental logic and committed checkpoint updates packages/envio/src/InMemoryStore.res	writeBatch now filters entity changes to those with checkpoint > committedCheckpointId; after persistence the store explicitly resets rawEvents, effects, rollback, updates committedCheckpointId from the batch last checkpoint, and truncates or recreates per-entity tables bounded by keepLatestChangesLimit.
Rollback diff application with committed checkpoint threading packages/envio/src/InMemoryStore.res	prepareRollbackDiff explicitly reinitializes rawEvents, entities, effects; rollback Delete/Set paths now call InMemoryTable.Entity.set with ~committedCheckpointId=inMemoryStore.committedCheckpointId.
Application wiring and incremental update validation packages/envio/src/UserContext.res, scenarios/test_codegen/test/helpers/MockIndexer.res, scenarios/test_codegen/test/WriteRead_test.res	UserContext set/deleteUnsafe forward committedCheckpointId to InMemoryTable.Entity.set; mock indexer passes committedCheckpointId; new test verifies committed entities persist across batches and history checkpoints are recorded correctly.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• enviodev/hyperindex#1247: Modifies InMemoryStore.writeBatch / prepareRollbackDiff behavior; overlaps with this PR's reset and write-path changes.
• enviodev/hyperindex#1258: Related changes to entity history selection and InMemoryTable.Entity.set logic; current PR extends checkpoint filtering and threads committed checkpoint through entity set calls.
• enviodev/hyperindex#1244: Touches in-memory store reset/rollback refactor that this PR builds upon.

Suggested reviewers

• JonoPrest
• moose-code

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title directly and clearly summarizes the main change: preserving committed entities across batches without rewriting their history, which is the core objective of the PR.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

packages/envio/src/InMemoryStore.res (1)

44-44: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Typo: commitedCheckpointId should be committedCheckpointId.

"Committed" has double 't'. Since this is a new field being introduced, fixing the spelling now avoids propagating the typo across the codebase.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/envio/src/InMemoryStore.res` at line 44, Rename the misspelled
mutable field commitedCheckpointId to committedCheckpointId in InMemoryStore.res
and update every reference to that field (constructors, record literals, pattern
matches, getters/setters, and any modules accessing Internal.checkpointId) to
use the new spelling; ensure compilation by running the build/tests and fix any
remaining identifier usages or imports that referenced commitedCheckpointId.

🧹 Nitpick comments (1)

packages/envio/src/InMemoryStore.res (1)

169-175: 💤 Low value

Consider using reduce instead of ref for accumulation.

This is more idiomatic ReScript:

♻️ Suggested refactor

-    let totalLatestChanges = ref(0)
-    persistence.allEntities->Belt.Array.forEach(entityConfig => {
-      totalLatestChanges :=
-        totalLatestChanges.contents +
-        (inMemoryStore->getInMemTable(~entityConfig)).latestEntityChangeById->Utils.Dict.size
-    })
-    let keepLatestChanges = totalLatestChanges.contents < keepLatestChangesLimit
+    let totalLatestChanges = persistence.allEntities->Belt.Array.reduce(0, (acc, entityConfig) => {
+      acc + (inMemoryStore->getInMemTable(~entityConfig)).latestEntityChangeById->Utils.Dict.size
+    })
+    let keepLatestChanges = totalLatestChanges < keepLatestChangesLimit

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/envio/src/InMemoryStore.res` around lines 169 - 175, The
accumulation using a mutable ref (totalLatestChanges) is non-idiomatic — replace
the ref loop with a pure fold: use persistence.allEntities->Belt.Array.reduce to
sum each entity's
inMemoryStore->getInMemTable(~entityConfig).latestEntityChangeById->Utils.Dict.size
and then compare the resulting total to keepLatestChangesLimit to set
keepLatestChanges; remove totalLatestChanges ref and the forEach to keep code
functional and clearer.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@packages/envio/src/InMemoryStore.res`:
- Line 44: Rename the misspelled mutable field commitedCheckpointId to
committedCheckpointId in InMemoryStore.res and update every reference to that
field (constructors, record literals, pattern matches, getters/setters, and any
modules accessing Internal.checkpointId) to use the new spelling; ensure
compilation by running the build/tests and fix any remaining identifier usages
or imports that referenced commitedCheckpointId.

---

Nitpick comments:
In `@packages/envio/src/InMemoryStore.res`:
- Around line 169-175: The accumulation using a mutable ref (totalLatestChanges)
is non-idiomatic — replace the ref loop with a pure fold: use
persistence.allEntities->Belt.Array.reduce to sum each entity's
inMemoryStore->getInMemTable(~entityConfig).latestEntityChangeById->Utils.Dict.size
and then compare the resulting total to keepLatestChangesLimit to set
keepLatestChanges; remove totalLatestChanges ref and the forEach to keep code
functional and clearer.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 124767f1-b0f4-4af3-8d03-a03f96dc2a5a

📥 Commits

Reviewing files that changed from the base of the PR and between 386ff16 and 0e6a951.

📒 Files selected for processing (1)

• packages/envio/src/InMemoryStore.res