feat(EN-1550): idempotent mirror ingest on v2LogId

docs/technical/architecture/audit-vs-technical-state.md

@@ -93,17 +93,26 @@ manual repair, or future tooling. The cursor is wrong in **both** directions:
 - A cursor that advances **beyond** the highest audited `MirrorIngest.v2LogId`
   makes the worker skip source logs without any audit entry showing the missing
   proposals — silent under-ingestion.
-- A cursor that drops **below** the highest audited `v2LogId` is equally unsafe.
-  `Worker.processBatch` (`internal/application/mirror/worker.go`) reloads the
-  cursor and calls `FetchLogs(cursor, ...)`, which returns source logs strictly
-  *after* the cursor. `TranslateBatch` then emits fresh `MirrorIngest` orders for
-  logs that were already ingested, and `processMirrorCreatedTransaction`
-  (`internal/domain/processing/processor_mirror.go`) reapplies every posting with
-  `force=true`, overwriting the transaction state and re-adding the amounts to the
-  volume pair via `applyPosting`. There is **no deduplication on `v2LogId`** and
-  volume mutation is additive, so a lowered cursor replays already-audited
-  transactions and applies their balance effects a second time — silent
-  double-application.
+- A cursor that drops **below** the highest audited `v2LogId` makes
+  `Worker.processBatch` (`internal/application/mirror/worker.go`) reload the
+  cursor and call `FetchLogs(cursor, ...)`, which returns source logs strictly
+  *after* the cursor. `TranslateBatch` then re-emits `MirrorIngest` orders for
+  logs that were already ingested. **This replay is now idempotent at the FSM
+  level (EN-1550).** `processMirrorIngest`
+  (`internal/domain/processing/processor_mirror.go`) records the highest applied
+  source id in `LedgerBoundaries.last_mirror_v2_log_id` and, *before* applying any
+  posting or mutating any state, skips any entry whose `v2LogId` is `<=` that
+  high-water mark, returning a deterministic no-op `(nil, nil)` — no postings
+  re-forced, no volume mutation, no fresh ledger log (`ProcessOrders` treats the
+  `(nil, nil)` outcome as "no log": no sequence id consumed, no audit-visible
+  `Log`). The boundary is a pure function of applied per-ledger state and v2 log
+  ids are 1-based and strictly increasing per source (`TranslateBatch`), so the
+  guard is FSM-deterministic and needs no new preload/coverage key (it lives
+  inside the already-covered boundaries). Historically (pre-EN-1550) a lowered
+  cursor caused silent **double-application**: postings were reapplied with
+  `force=true` and additive volume mutation with no dedup on `v2LogId`. Only the
+  **behind** direction is closed here; recovering a cursor that is *ahead* of the
+  true source head is worker-side source-head recovery and remains out of scope.
 
 Because it is only correct at one exact value, treat `MirrorCursor` as a
 persisted per-ledger projection (`ZonePerLedger` / `SubPLMirrorCursor`, in the
@@ -138,6 +147,20 @@ every node identically. Mirror recovery, repair tooling, or a checker pass
 that re-derives the expected cursor from the audited mirror-ingest logs must
 enforce the equality before the stored cursor can be trusted.
 
+`LedgerBoundaries.last_mirror_v2_log_id` (the EN-1550 idempotency high-water
+mark) does **not** add a new tamper vector and needs **no** new checker pass. It
+is derivable from the audited `MirrorIngest` orders — it is exactly the maximum
+applied `v2LogId`, the same quantity the (still-owed) `MirrorCursor` equality
+above re-derives — and it only *gates future application* by turning a replayed
+ingest into a no-op. Tampering it can only make the guard stricter (skip a not-yet
+-applied log, which the pending `MirrorCursor` equality check would already flag
+as under-ingestion) or looser (admit an already-applied `v2LogId`), and the
+looser case cannot double a balance on its own: the double-application it would
+re-enable is precisely what the same-proposal, audit-bound posting effects and
+the `MirrorCursor` equality already govern. It is not independently
+business-authoritative, so it rides on the mirror-cursor coverage rather than
+earning its own `compare*` pass.
+
 ## Readstore and Indexes
 
 The index **registry** is business-visible and is already verified by the

internal/domain/processing/processor.go

@@ -253,6 +253,24 @@ func (p *RequestProcessor) ProcessOrders(orders []*raftcmdpb.Order, scopeFactory
 			return nil, err
 		}
 
+		// No-log outcome: a processor may deterministically decide the order
+		// produces no fresh ledger log while still succeeding (payload==nil,
+		// err==nil). The only such case today is an idempotent mirror replay
+		// (processMirrorIngest guards on LastMirrorV2LogId), which mutates
+		// nothing and must leave no audit-visible log. Skip the log slot
+		// entirely — do NOT consume a sequence id, absorb into the sink, or
+		// append a degenerate Log{Payload:nil}. The nil slot mirrors the
+		// idempotency-replay ReferenceSequence path and is skipped identically
+		// by WriteSet.Merge (GetCreatedLog()==nil) and checkCloseChapter.
+		//
+		// Any overlay staged by such an order is dropped without Commit(): a
+		// no-log outcome makes no persistent mutation, so there is nothing to
+		// flush. (Mirror ingest declares no skippable_reasons, so no overlay is
+		// created here in practice.)
+		if payload == nil {
+			continue
+		}
+
 		if overlay != nil {
 			if err := overlay.Commit(); err != nil {
 				// Coverage-miss surfaced by a staged Delete (invariant #6):

internal/domain/processing/processor_mirror.go

@@ -27,10 +27,6 @@ func processMirrorIngest(ledger string, order *raftcmdpb.MirrorIngestOrder, ctx
 	if info.GetMode() != commonpb.LedgerMode_LEDGER_MODE_MIRROR {
 		return nil, &domain.ErrLedgerNotInMirrorMode{Name: ledger}
 	}
-	// Re-touch ledger info so it enters the Merge buffer and gets propagated
-	// back to Gen0 on commit. Without this, ledger info is evicted after two
-	// cache rotations because mirror proposals bypass the admission preloader.
-	s.Ledgers().Put(domain.LedgerKey{Name: ledger}, info)
 
 	boundariesReader, loadErr := loadBoundaries(s, ledger)
 	if loadErr != nil {
@@ -39,15 +35,48 @@ func processMirrorIngest(ledger string, order *raftcmdpb.MirrorIngestOrder, ctx
 
 	boundaries := boundariesReader.Mutate()
 
-	// Stage per-apply context fields for child handlers.
-	ctx.Boundaries = boundaries
-	ctx.LedgerInfo = info
-
 	entry := order.GetEntry()
 	if entry == nil {
 		return nil, &domain.ErrLedgerNotInMirrorMode{Name: ledger}
 	}
 
+	// Idempotent replay guard — evaluated BEFORE any mutation (no ledger
+	// re-touch, no cache write) so a replayed ingest is a truly side-effect-free
+	// no-op. The mirror worker resumes from the MirrorCursor projection; if that
+	// cursor is tampered (or rolled back) to a lower value, the worker re-fetches
+	// and re-proposes source logs that were already applied here, force-adding
+	// their postings a second time and doubling balances (flemzord, #1581).
+	// LastMirrorV2LogId is the highest source v2LogId already applied to this
+	// ledger — a pure function of applied per-ledger state. v2 log IDs are
+	// 1-based and strictly increasing per source (see adapter/v2/translator.go:
+	// TranslateBatch), so an entry whose v2LogId is <= the recorded high-water
+	// mark has already been applied and re-applying it must be a deterministic
+	// no-op.
+	//
+	// This is an EXPECTED stale/replayed-ingest case, not an impossible-by-design
+	// branch, so a soft skip (return nil, nil — no log payload, no mutation) is
+	// the correct outcome per invariant #7's expected-vs-impossible distinction;
+	// failing loud (assert.Unreachable) is reserved for branches reachable only
+	// when an invariant is already broken, which replay is not. ProcessOrders
+	// treats the (nil, nil) return as a no-log outcome: no sequence id, no audit
+	// log, no sink absorb.
+	//
+	// First-ingest case: LastMirrorV2LogId defaults to 0, so the first real
+	// v2LogId (>= 1) is strictly greater and applies normally.
+	v2LogID := entry.GetV2LogId()
+	if v2LogID != 0 && v2LogID <= boundaries.GetLastMirrorV2LogId() {
+		return nil, nil
+	}
+
+	// Re-touch ledger info so it enters the Merge buffer and gets propagated
+	// back to Gen0 on commit. Without this, ledger info is evicted after two
+	// cache rotations because mirror proposals bypass the admission preloader.
+	s.Ledgers().Put(domain.LedgerKey{Name: ledger}, info)
+
+	// Stage per-apply context fields for child handlers.
+	ctx.Boundaries = boundaries
+	ctx.LedgerInfo = info
+
 	var logPayload *commonpb.LedgerLogPayload
 
 	switch data := entry.GetData().(type) {
@@ -93,6 +122,14 @@ func processMirrorIngest(ledger string, order *raftcmdpb.MirrorIngestOrder, ctx
 	// Assign per-ledger log ID and advance boundaries
 	nextLogID := boundaries.GetNextLogId()
 	boundaries.NextLogId = nextLogID + 1
+	// Advance the idempotent-replay high-water mark. Set once here regardless of
+	// the inner ingest kind, because v2LogId lives on the wrapping MirrorLogEntry
+	// and applies to every kind (CreatedTransaction, SavedMetadata,
+	// DeletedMetadata, RevertedTransaction, FillGap). Guarded by v2LogID != 0 so
+	// an entry that somehow carries no source id never rewinds the mark.
+	if v2LogID != 0 {
+		boundaries.LastMirrorV2LogId = v2LogID
+	}
 	s.Boundaries().Put(domain.LedgerKey{Name: ledger}, boundaries)
 
 	return &commonpb.LogPayload{

internal/domain/processing/processor_mirror_test.go

@@ -168,6 +168,161 @@ func TestMirrorIngest_CreatedTransaction(t *testing.T) {
 	require.Equal(t, uint64(43), putBoundaries.GetNextTransactionId())
 }
 
+// mirrorCreatedTxOrder builds a MirrorIngest order for a created transaction
+// carrying the given v2LogId. Shared by the idempotency tests so they exercise
+// the exact wrapper shape (v2LogId on MirrorLogEntry) the guard reads.
+func mirrorCreatedTxOrder(ledger string, v2LogID, txID uint64) *raftcmdpb.Order {
+	return &raftcmdpb.Order{
+		Type: &raftcmdpb.Order_LedgerScoped{
+			LedgerScoped: &raftcmdpb.LedgerScopedOrder{
+				Ledger: ledger,
+				Payload: &raftcmdpb.LedgerScopedOrder_MirrorIngest{
+					MirrorIngest: &raftcmdpb.MirrorIngestOrder{Entry: &raftcmdpb.MirrorLogEntry{
+						V2LogId: v2LogID,
+						Data: &raftcmdpb.MirrorLogEntry_CreatedTransaction{
+							CreatedTransaction: &raftcmdpb.MirrorCreatedTransaction{
+								TransactionId: txID,
+								Postings: []*commonpb.Posting{{
+									Source:      "world",
+									Destination: "users:001",
+									Amount:      commonpb.NewUint256FromUint64(500),
+									Asset:       "USD/2",
+								}},
+							},
+						},
+					}},
+				},
+			},
+		},
+	}
+}
+
+// TestMirrorIngest_ReplayIsNoOp pins the EN-1550 idempotency guard: an ingest
+// whose v2LogId equals the recorded high-water mark (LastMirrorV2LogId) has
+// already been applied, so re-applying it is a deterministic side-effect-free
+// no-op — no postings applied (no volume writes → balances not doubled), no
+// ledger re-touch, no boundary advance, and (nil, nil) returned so ProcessOrders
+// emits no log.
+func TestMirrorIngest_ReplayIsNoOp(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockStore := NewMockScope(ctrl)
+	processor, err := NewRequestProcessor(nil, 0)
+	require.NoError(t, err)
+
+	// v2LogId 7 was already applied: LastMirrorV2LogId == 7.
+	boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 43, NextLogId: 5, LastMirrorV2LogId: 7}
+	ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR}
+
+	expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes()
+
+	boundariesStub := setupBoundariesStub(mockStore)
+	boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil)
+	// The guard runs before any mutation: no boundary Put on replay.
+	boundariesStub.onPut(func(_ domain.LedgerKey, _ *raftcmdpb.LedgerBoundaries) {
+		t.Errorf("Boundaries().Put must not be called on an idempotent replay")
+	})
+	// No ledger re-touch on replay (guarded before s.Ledgers().Put).
+	ledgersStub := setupLedgersStub(mockStore)
+	ledgersStub.onPut(func(_ domain.LedgerKey, _ *commonpb.LedgerInfo) {
+		t.Errorf("Ledgers().Put must not be called on an idempotent replay")
+	})
+	// No posting applied → no volume writes → balances cannot double.
+	volumesStub := setupVolumesStub(mockStore)
+	volumesStub.onPut(func(_ domain.VolumeKey, _ *raftcmdpb.VolumePair) {
+		t.Errorf("Volumes().Put must not be called on an idempotent replay")
+	})
+
+	// Replay v2LogId 7 (already applied).
+	result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 7, 42), mockStore)
+	require.NoError(t, err)
+	require.Nil(t, result, "replayed ingest must produce no log payload")
+}
+
+// TestMirrorIngest_LowerV2LogIdSkipped pins that an OLDER v2LogId arriving after
+// a higher one (e.g. a tampered/rolled-back MirrorCursor makes the worker
+// re-emit past logs) is skipped as a no-op.
+func TestMirrorIngest_LowerV2LogIdSkipped(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockStore := NewMockScope(ctrl)
+	processor, err := NewRequestProcessor(nil, 0)
+	require.NoError(t, err)
+
+	boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 43, NextLogId: 5, LastMirrorV2LogId: 10}
+	ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR}
+
+	expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes()
+
+	boundariesStub := setupBoundariesStub(mockStore)
+	boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil)
+	boundariesStub.onPut(func(_ domain.LedgerKey, _ *raftcmdpb.LedgerBoundaries) {
+		t.Errorf("Boundaries().Put must not be called for a stale (lower) v2LogId")
+	})
+	volumesStub := setupVolumesStub(mockStore)
+	volumesStub.onPut(func(_ domain.VolumeKey, _ *raftcmdpb.VolumePair) {
+		t.Errorf("Volumes().Put must not be called for a stale (lower) v2LogId")
+	})
+
+	// v2LogId 6 is older than the applied high-water mark 10.
+	result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 6, 99), mockStore)
+	require.NoError(t, err)
+	require.Nil(t, result)
+}
+
+// TestMirrorIngest_AdvancesLastMirrorV2LogId pins that a forward ingest applies
+// and records its v2LogId as the new high-water mark on the boundaries it writes.
+func TestMirrorIngest_AdvancesLastMirrorV2LogId(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockStore := NewMockScope(ctrl)
+	processor, err := NewRequestProcessor(nil, 0)
+	require.NoError(t, err)
+
+	now := &commonpb.Timestamp{Data: 1234567890}
+	// Previously applied up to v2LogId 3; the new entry (4) is strictly greater.
+	boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 1, NextLogId: 1, LastMirrorV2LogId: 3}
+	ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR}
+
+	var putBoundaries *raftcmdpb.LedgerBoundaries
+
+	expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes()
+	expectPutLedger(t, mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo)
+	mockStore.EXPECT().GetDate().Return(now.AsReader()).AnyTimes()
+	mockStore.EXPECT().GetNextSequenceID().Return(uint64(100))
+	mockStore.EXPECT().GetCurrentOpenChapter().Return(nil, false)
+
+	boundariesStub := setupBoundariesStub(mockStore)
+	boundariesStub.expectGet(domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil)
+	boundariesStub.onPut(func(_ domain.LedgerKey, b *raftcmdpb.LedgerBoundaries) { putBoundaries = b })
+
+	zeroVol := &raftcmdpb.VolumePair{
+		Input:  commonpb.NewUint256FromUint64(0),
+		Output: commonpb.NewUint256FromUint64(0),
+	}
+	volumes := setupVolumesStub(mockStore)
+	volumes.expectGet(domain.NewVolumeKey("mirror-ledger", "world", "USD/2"), zeroVol.AsReader(), nil)
+	volumes.expectGet(domain.NewVolumeKey("mirror-ledger", "users:001", "USD/2"), zeroVol.AsReader(), nil)
+	expectPutTransactionState(t, mockStore, domain.TransactionKey{LedgerName: "mirror-ledger", ID: 42}, nil)
+
+	result, err := processor.ProcessOrder(mirrorCreatedTxOrder("mirror-ledger", 4, 42), mockStore)
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	require.NotNil(t, putBoundaries)
+	require.Equal(t, uint64(4), putBoundaries.GetLastMirrorV2LogId(),
+		"applied ingest must advance the v2LogId high-water mark")
+}
+
 func TestMirrorIngest_NotMirrorMode(t *testing.T) {
 	t.Parallel()
 

internal/domain/processing/processor_test.go

@@ -440,3 +440,62 @@ func TestProcessOrders_OrdersResultEmpty(t *testing.T) {
 	require.Equal(t, uint64(0), response.MinLogSequence)
 	require.Equal(t, uint64(0), response.MaxLogSequence)
 }
+
+// TestProcessOrders_MirrorReplayEmitsNoLog pins the EN-1550 no-log outcome at
+// the batch level: an idempotent mirror replay (v2LogId <= LastMirrorV2LogId)
+// returns (nil, nil) from processMirrorIngest, so ProcessOrders must NOT consume
+// a sequence id, NOT append a degenerate Log{Payload:nil}, and NOT record a
+// CreatedLog — the per-order slot stays nil (skipped by WriteSet.Merge exactly
+// like an idempotency-replay ReferenceSequence). Without the guard, replaying an
+// already-applied source log would re-force its postings and double balances
+// (#1581). The sink is asserted to receive zero Absorb calls.
+func TestProcessOrders_MirrorReplayEmitsNoLog(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockStore := NewMockScope(ctrl)
+	processor, err := NewRequestProcessor(nil, 0)
+	require.NoError(t, err)
+
+	// v2LogId 7 already applied.
+	boundaries := &raftcmdpb.LedgerBoundaries{NextTransactionId: 43, NextLogId: 5, LastMirrorV2LogId: 7}
+	ledgerInfo := &commonpb.LedgerInfo{Name: "mirror-ledger", Mode: commonpb.LedgerMode_LEDGER_MODE_MIRROR}
+
+	expectGetLedger(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, ledgerInfo.AsReader(), nil).AnyTimes()
+	expectGetBoundaries(mockStore, domain.LedgerKey{Name: "mirror-ledger"}, boundaries.AsReader(), nil).AnyTimes()
+
+	// A no-op must consume no proposal-level sequence id: IncrementNextSequenceID
+	// is registered with Times(0) so any call fails the test.
+	mockStore.EXPECT().IncrementNextSequenceID().Times(0)
+
+	// A replay of v2LogId 7 (already applied) followed by an even older one (5).
+	orders := []*raftcmdpb.Order{
+		mirrorCreatedTxOrder("mirror-ledger", 7, 42),
+		mirrorCreatedTxOrder("mirror-ledger", 5, 41),
+	}
+
+	sink := &countingSink{}
+	response, err := processor.ProcessOrders(orders, mockFactory(mockStore), sink)
+	require.NoError(t, err)
+	require.NotNil(t, response)
+
+	// Two order slots exist (index alignment preserved) but both are nil — no
+	// log produced for either replay.
+	require.Len(t, response.Logs, 2)
+	require.Nil(t, response.Logs[0].GetCreatedLog())
+	require.Nil(t, response.Logs[1].GetCreatedLog())
+
+	// No CreatedLogs, no sequence range, no sink absorb.
+	require.Empty(t, response.CreatedLogs)
+	require.Equal(t, uint64(0), response.MinLogSequence)
+	require.Equal(t, uint64(0), response.MaxLogSequence)
+	require.Equal(t, 0, sink.count)
+}
+
+// countingSink records how many (order, log) pairs it absorbed so a test can
+// assert a no-log order never reaches the sink.
+type countingSink struct{ count int }
+
+func (s *countingSink) Absorb(_ *raftcmdpb.Order, _ *commonpb.Log) { s.count++ }