fix: repair missing DuckDB mirror sessions

Incremental DuckDB pushes could prune a cached fingerprint after discovering a missing mirror row, then advance the watermark without adding the unchanged local session back to the push candidate set. That left the mirror missing the row until a full push or later local modification.

Treat pruned fingerprints that still have in-scope local sessions as repair candidates before fingerprint filtering, so the normal push path reinserts the session and dependents. The regression test deletes a mirrored row after a successful full push and verifies the next incremental push restores it.

Author: cpcloud

internal/duckdb/sync.go

@@ -270,13 +270,11 @@ func (s *Sync) Push(
 	if err != nil {
 		return result, fmt.Errorf("listing local sessions: %w", err)
 	}
-	result.Diagnostics.LocalSessions = countPushSessions(allLocalSessions)
-	sessionFingerprints, err := s.sessionFingerprints(ctx, sessions)
-	if err != nil {
-		return result, err
+	allLocalSessionByID := make(map[string]db.Session, len(allLocalSessions))
+	for _, sess := range allLocalSessions {
+		allLocalSessionByID[sess.ID] = sess
 	}
-	candidateSessions := append([]db.Session(nil), sessions...)
-	result.Diagnostics.CandidateSessions = countPushSessions(candidateSessions)
+	result.Diagnostics.LocalSessions = countPushSessions(allLocalSessions)
 	priorFingerprints := map[string]string{}
 	if !full {
 		priorFingerprints, err = readSyncFingerprintsWithKey(
@@ -305,7 +303,20 @@ func (s *Sync) Push(
 	}
 	result.Diagnostics.DeletedStaleSessions = len(staleIDs)
 	if !full {
-		pruneMissingMirrorFingerprints(priorFingerprints, mirrorSessionIDs)
+		missingMirrorIDs := pruneMissingMirrorFingerprints(
+			priorFingerprints, mirrorSessionIDs,
+		)
+		sessions = appendMissingMirrorRepairCandidates(
+			sessions, sessionByID, allLocalSessionByID, missingMirrorIDs,
+		)
+	}
+	sessionFingerprints, err := s.sessionFingerprints(ctx, sessions)
+	if err != nil {
+		return result, err
+	}
+	candidateSessions := append([]db.Session(nil), sessions...)
+	result.Diagnostics.CandidateSessions = countPushSessions(candidateSessions)
+	if !full {
 		sessions = filterUnchangedSessions(sessions, priorFingerprints, sessionFingerprints)
 		result.Diagnostics.SkippedUnchangedSessions = skippedPushSessions(
 			candidateSessions, sessions,
@@ -884,12 +895,36 @@ func filterUnchangedSessions(
 func pruneMissingMirrorFingerprints(
 	priorFingerprints map[string]string,
 	mirrorSessionIDs map[string]bool,
-) {
+) []string {
+	var missing []string
 	for id := range priorFingerprints {
 		if !mirrorSessionIDs[id] {
 			delete(priorFingerprints, id)
+			missing = append(missing, id)
+		}
+	}
+	sort.Strings(missing)
+	return missing
+}
+
+func appendMissingMirrorRepairCandidates(
+	sessions []db.Session,
+	sessionByID map[string]db.Session,
+	allLocalSessionByID map[string]db.Session,
+	missingMirrorIDs []string,
+) []db.Session {
+	for _, id := range missingMirrorIDs {
+		sess, ok := allLocalSessionByID[id]
+		if !ok {
+			continue
+		}
+		if _, ok := sessionByID[id]; ok {
+			continue
 		}
+		sessionByID[id] = sess
+		sessions = append(sessions, sess)
 	}
+	return sessions
 }
 
 func previousLocalSyncTimestamp(value string) (string, error) {

internal/duckdb/sync_fastpath_test.go

@@ -392,6 +392,36 @@ func TestSyncCheckpointFailureAfterHardDeleteDoesNotKeepStaleFingerprint(t *test
 	assert.NotContains(t, fingerprints, fixture.betaID)
 }
 
+func TestSyncMissingMirrorRowWithUnchangedLocalSessionIsRepaired(t *testing.T) {
+	ctx := context.Background()
+	local := newLocalDB(t)
+	fixture := seedDuckDBSyncFixture(t, local)
+	syncer := newInMemoryTestSync(t, local, SyncOptions{})
+	policy := &checkpointSpy{}
+	syncer.maintenance = policy
+
+	first, err := syncer.Push(ctx, true, nil)
+	require.NoError(t, err)
+	require.Equal(t, 2, first.SessionsPushed)
+	assert.Equal(t, 1, policy.calls)
+	require.NoError(t, syncer.withDuckTx(ctx, "test delete mirror session", func(tx *sql.Tx) error {
+		return syncer.deleteMirrorSession(ctx, tx, fixture.betaID)
+	}))
+	assertDuckDBCountWhere(t, syncer.DB(), "sessions", "id = ?", fixture.betaID, 0)
+
+	second, err := syncer.Push(ctx, false, nil)
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, second.SessionsPushed)
+	assert.Equal(t, 1, second.MessagesPushed)
+	assert.Equal(t, 2, policy.calls)
+	assertDuckDBCountWhere(t, syncer.DB(), "sessions", "id = ?", fixture.betaID, 1)
+	assertDuckDBCountWhere(t, syncer.DB(), "messages", "session_id = ?", fixture.betaID, 1)
+	fingerprints, err := readSyncFingerprintsWithKey(local, lastPushBoundaryStateKey)
+	require.NoError(t, err)
+	assert.Contains(t, fingerprints, fixture.betaID)
+}
+
 func TestSyncCheckpointPolicySkipsRemoteQuackTargets(t *testing.T) {
 	ctx := context.Background()
 	local := newLocalDB(t)