fix: surface reconciliation errors for assertion persistence and finalizer ordering (#2034)

* fix: surface reconciliation errors for assertion persistence and finalizer ordering

- handlePKFailure now captures assertionRepo.Update errors and returns a combined
  error wrapping both the PK failure and the persistence failure instead of silently
  discarding with _ =
- markFinalized reorders operations so snapshots are marked FINAL before the run is
  persisted as FINALIZED, preventing a state where the run is FINALIZED but its
  snapshots are not yet FINAL

* fix: resolve business run_id to surrogate PK in MarkRunSnapshotsFinal

settlement_snapshot.run_id is a FK to settlement_run.id (the surrogate PK),
not settlement_run.run_id (the business identifier). The previous implementation
passed the UUID directly to the SQL query without resolving the FK-correct
surrogate ID, resulting in a silent 0-row update when called with a business UUID.

MarkRunSnapshotsFinal now resolves settlement_run.run_id -> settlement_run.id
internally before updating, consistent with the integration test repo's approach.
Updated the persistence integration test to create snapshots with the surrogate ID
(FK correct) and call MarkRunSnapshotsFinal with the business UUID.

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

services/reconciliation/adapters/persistence/settlement_snapshot_repository.go

@@ -3,6 +3,7 @@ package persistence
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/google/uuid"
@@ -138,14 +139,20 @@ func (r *SettlementSnapshotRepository) DeleteByRunID(ctx context.Context, runID
 }
 
 // MarkRunSnapshotsFinal updates all snapshots for a run to include settlement_type=FINAL in their attributes.
+// runID is the business identifier (settlement_run.run_id); this method resolves it to the surrogate PK
+// (settlement_run.id) before updating, since settlement_snapshot.run_id is a FK to settlement_run.id.
 func (r *SettlementSnapshotRepository) MarkRunSnapshotsFinal(ctx context.Context, runID uuid.UUID) error {
 	return r.withTenantTransaction(ctx, func(tx *gorm.DB) error {
+		var runEntity SettlementRunEntity
+		if err := tx.Select("id").Where("run_id = ?", runID).First(&runEntity).Error; err != nil {
+			return fmt.Errorf("resolving surrogate ID for run %s: %w", runID, err)
+		}
 		// Use CASE to handle NULL/JSON-null attributes vs existing JSONB objects.
 		// CockroachDB's || operator requires both operands to be JSONB objects,
 		// so we must replace NULL/json-null with an empty object before merging.
 		return tx.Exec(
 			`UPDATE "settlement_snapshot" SET "attributes" = CASE WHEN "attributes" IS NULL OR "attributes"::text = 'null' THEN '{"settlement_type":"FINAL"}'::jsonb ELSE "attributes" || '{"settlement_type":"FINAL"}'::jsonb END WHERE "run_id" = ?`,
-			runID,
+			runEntity.ID,
 		).Error
 	})
 }

services/reconciliation/adapters/persistence/settlement_snapshot_repository_test.go

@@ -16,8 +16,23 @@ import (
 	"gorm.io/gorm"
 )
 
-// seedSettlementRun creates a settlement_run record and returns the surrogate ID.
+// seedRunIDs holds the two distinct identifiers for a seeded settlement_run row.
+type seedRunIDs struct {
+	// SurrogateID is the settlement_run.id (DB surrogate PK); used as the FK target in settlement_snapshot.run_id.
+	SurrogateID uuid.UUID
+	// BusinessID is the settlement_run.run_id (business identifier); passed to MarkRunSnapshotsFinal and similar methods.
+	BusinessID uuid.UUID
+}
+
+// seedSettlementRun creates a settlement_run record and returns the surrogate ID (settlement_run.id).
+// For tests that also need the business run_id, use seedSettlementRunFull.
 func seedSettlementRun(t *testing.T, ctx context.Context, db *gorm.DB) uuid.UUID {
+	t.Helper()
+	return seedSettlementRunFull(t, ctx, db).SurrogateID
+}
+
+// seedSettlementRunFull creates a settlement_run record and returns both the surrogate ID and the business run_id.
+func seedSettlementRunFull(t *testing.T, ctx context.Context, db *gorm.DB) seedRunIDs {
 	t.Helper()
 	tid := tenant.TenantID("test-tenant-01")
 	quoted := fmt.Sprintf("%q", tid.SchemaName())
@@ -30,7 +45,7 @@ func seedSettlementRun(t *testing.T, ctx context.Context, db *gorm.DB) uuid.UUID
 		surrogateID, runID,
 	).Error
 	require.NoError(t, err)
-	return surrogateID
+	return seedRunIDs{SurrogateID: surrogateID, BusinessID: runID}
 }
 
 func newTestSnapshot(t *testing.T, runID uuid.UUID) *domain.SettlementSnapshot {
@@ -234,29 +249,33 @@ func TestSettlementSnapshotRepository_MarkRunSnapshotsFinal(t *testing.T) {
 	repo := persistence.NewSettlementSnapshotRepository(db)
 	ctx := tenantCtx()
 
-	runID := seedSettlementRun(t, ctx, db)
+	// Use seedSettlementRunFull so we have both IDs:
+	//   SurrogateID → stored in settlement_snapshot.run_id (FK target)
+	//   BusinessID  → passed to MarkRunSnapshotsFinal (business identifier)
+	run := seedSettlementRunFull(t, ctx, db)
 
-	// Create snapshots with various attributes
-	s1 := newTestSnapshot(t, runID)
+	// Create snapshots with the surrogate ID so the FK constraint is satisfied.
+	s1 := newTestSnapshot(t, run.SurrogateID)
 	s1.Attributes = map[string]string{"bucket": "default"}
 	require.NoError(t, repo.Create(ctx, s1))
 
-	s2 := newTestSnapshot(t, runID)
+	s2 := newTestSnapshot(t, run.SurrogateID)
 	s2.AccountID = "ACC-002"
 	s2.Attributes = nil // Test nil attributes case
 	require.NoError(t, repo.Create(ctx, s2))
 
-	s3 := newTestSnapshot(t, runID)
+	s3 := newTestSnapshot(t, run.SurrogateID)
 	s3.AccountID = "ACC-003"
 	s3.Attributes = map[string]string{"region": "eu-west-1"}
 	require.NoError(t, repo.Create(ctx, s3))
 
-	// Mark all snapshots as FINAL
-	err := repo.MarkRunSnapshotsFinal(ctx, runID)
+	// Mark all snapshots as FINAL using the business run_id (not the surrogate PK).
+	// MarkRunSnapshotsFinal resolves the business ID to the surrogate PK internally.
+	err := repo.MarkRunSnapshotsFinal(ctx, run.BusinessID)
 	require.NoError(t, err)
 
 	// Verify all snapshots have settlement_type=FINAL
-	found, err := repo.FindByRunID(ctx, runID)
+	found, err := repo.FindByRunID(ctx, run.SurrogateID)
 	require.NoError(t, err)
 	assert.Len(t, found, 3)
 	for _, snap := range found {

services/reconciliation/service/balance_assertor.go

@@ -166,9 +166,12 @@ func (ba *BalanceAssertor) handlePKFailure(ctx context.Context, assertion *domai
 	if failErr := assertion.Fail(decimal.Zero, failReason); failErr != nil {
 		ba.logger.Error("failed to mark assertion as failed", "error", failErr)
 	}
-	_ = ba.assertionRepo.Update(ctx, assertion)
+	retErr := fmt.Errorf("querying position keeping: %w", pkErr)
+	if updateErr := ba.assertionRepo.Update(ctx, assertion); updateErr != nil {
+		retErr = fmt.Errorf("%w; persisting FAILED assertion: %w", retErr, updateErr)
+	}
 	observability.BalanceAssertionTotal.WithLabelValues("FAILED", scope.String()).Inc()
-	return &AssertBalanceResult{Assertion: assertion}, fmt.Errorf("querying position keeping: %w", pkErr)
+	return &AssertBalanceResult{Assertion: assertion}, retErr
 }
 
 // handleBalanced records a PASSED assertion when debits equal credits.

services/reconciliation/service/balance_assertor_test.go

@@ -360,6 +360,33 @@ func TestExecuteBalanceAssertion_PKClientError(t *testing.T) {
 	assert.Equal(t, domain.AssertionStatusFailed, result.Assertion.Status)
 }
 
+func TestExecuteBalanceAssertion_PKClientError_WithRepoUpdateError(t *testing.T) {
+	repo := newMockAssertionRepo()
+	repo.updateErr = errors.New("db write failed")
+	pkClient := &mockPKClient{
+		err: errors.New("connection refused"),
+	}
+
+	assertor := NewBalanceAssertor(repo, nil, pkClient, nil, nil, testLogger())
+
+	result, err := assertor.ExecuteBalanceAssertion(context.Background(), AssertBalanceRequest{
+		AccountID:       "ACC-001",
+		InstrumentCode:  "GBP",
+		Expression:      "total_debits == total_credits",
+		ExpectedBalance: decimal.Zero,
+		Scope:           domain.AssertionScopePositionLedger,
+		CallerRole:      CallerRoleTenantAdmin,
+	})
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "querying position keeping")
+	assert.Contains(t, err.Error(), "persisting FAILED assertion")
+
+	// Result is still returned with the failed assertion
+	require.NotNil(t, result)
+	assert.Equal(t, domain.AssertionStatusFailed, result.Assertion.Status)
+}
+
 func TestExecuteBalanceAssertion_TrendTracking(t *testing.T) {
 	trendRepo := newMockTrendRepo()
 	publisher := &mockEventPublisher{}

services/reconciliation/service/finalizer.go

@@ -214,19 +214,20 @@ func (f *SettlementFinalizer) checkPendingOperations(ctx context.Context, run *d
 }
 
 // markFinalized transitions the run to FINALIZED and marks snapshots as FINAL.
+// Snapshots are marked FINAL before the run is persisted as FINALIZED to prevent
+// a state where the run appears FINALIZED but its snapshots are not yet FINAL.
 func (f *SettlementFinalizer) markFinalized(ctx context.Context, run *domain.SettlementRun) error {
+	if err := f.snapRepo.MarkRunSnapshotsFinal(ctx, run.RunID); err != nil {
+		return fmt.Errorf("marking snapshots FINAL for run %s: %w", run.RunID, err)
+	}
+
 	if err := run.Finalize(); err != nil {
 		return fmt.Errorf("transitioning run %s to FINALIZED: %w", run.RunID, err)
 	}
 	if err := f.runRepo.Update(ctx, run); err != nil {
 		return fmt.Errorf("persisting FINALIZED state for run %s: %w", run.RunID, err)
 	}
 
-	if err := f.snapRepo.MarkRunSnapshotsFinal(ctx, run.RunID); err != nil {
-		f.logger.WarnContext(ctx, "failed to mark snapshots as FINAL",
-			"run_id", run.RunID, "error", err)
-	}
-
 	return nil
 }
 

services/reconciliation/service/finalizer_test.go

@@ -395,7 +395,7 @@ func TestFinalizeSettlement_NilLockClient(t *testing.T) {
 	assert.Equal(t, domain.RunStatusFinalized, finalizedRun.Status)
 }
 
-func TestFinalizeSettlement_SnapshotMarkFailureNonFatal(t *testing.T) {
+func TestFinalizeSettlement_SnapshotMarkFailureBlocksFinalization(t *testing.T) {
 	runRepo := newMockRunRepo()
 	snapRepo := &mockFinalSnapshotRepo{markErr: errors.New("snapshot update failed")}
 	publisher := &mockFinalityPublisher{}
@@ -405,12 +405,13 @@ func TestFinalizeSettlement_SnapshotMarkFailureNonFatal(t *testing.T) {
 
 	finalizer := NewSettlementFinalizer(runRepo, snapRepo, nil, publisher, nil)
 	err := finalizer.FinalizeSettlement(serviceCtx(), run.RunID)
-	// Should succeed even if snapshot marking fails (non-fatal)
-	require.NoError(t, err)
+	// Should fail if snapshot marking fails - run must not be FINALIZED with non-FINAL snapshots
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "marking snapshots FINAL")
 
-	// Run should still be FINALIZED
-	finalizedRun, _ := runRepo.FindByID(context.Background(), run.RunID)
-	assert.Equal(t, domain.RunStatusFinalized, finalizedRun.Status)
+	// Run should NOT be FINALIZED since snapshots couldn't be marked FINAL
+	unchangedRun, _ := runRepo.FindByID(context.Background(), run.RunID)
+	assert.NotEqual(t, domain.RunStatusFinalized, unchangedRun.Status)
 }
 
 func TestFinalizeSettlement_PublisherFailureNonFatal(t *testing.T) {