fix: prune DuckDB fingerprints for missing mirror rows

A checkpoint failure can happen after hard-deleted DuckDB rows have already been removed but before the push boundary fingerprints are finalized. On the next retry those rows are no longer reported as stale, so old cached fingerprints could be copied forward and later suppress a restored or missing-row session push.

Delete reconciliation now returns the mirror session IDs that remain after stale cleanup, and incremental skip filtering runs only after cached fingerprints have been pruned against that mirror state. A regression covers the checkpoint-failure retry path without advancing the watermark.

Author: cpcloud

internal/duckdb/push.go

@@ -316,42 +316,45 @@ func (s *Sync) replaceSessionDependents(
 func (s *Sync) deleteHardDeletedMirrorSessions(
 	ctx context.Context, tx *sql.Tx, localSessions []db.Session,
 	machine string, projects, excludeProjects []string,
-) ([]string, error) {
+) ([]string, map[string]bool, error) {
 	localIDs := make(map[string]bool, len(localSessions))
 	for _, sess := range localSessions {
 		localIDs[sess.ID] = true
 	}
+	mirrorIDs := make(map[string]bool)
 	rows, err := tx.QueryContext(ctx,
 		`SELECT id, project FROM sessions WHERE machine = ?`,
 		machine,
 	)
 	if err != nil {
-		return nil, fmt.Errorf("listing duckdb sessions for deletion reconciliation: %w", err)
+		return nil, nil, fmt.Errorf("listing duckdb sessions for deletion reconciliation: %w", err)
 	}
 	defer rows.Close()
 	var stale []string
 	for rows.Next() {
 		var id, project string
 		if err := rows.Scan(&id, &project); err != nil {
-			return nil, fmt.Errorf("scanning duckdb session for deletion reconciliation: %w", err)
+			return nil, nil, fmt.Errorf("scanning duckdb session for deletion reconciliation: %w", err)
 		}
+		mirrorIDs[id] = true
 		if !projectInSyncScope(project, projects, excludeProjects) {
 			continue
 		}
 		if !localIDs[id] {
 			stale = append(stale, id)
+			delete(mirrorIDs, id)
 		}
 	}
 	if err := rows.Err(); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	sort.Strings(stale)
 	for _, id := range stale {
 		if err := s.deleteMirrorSession(ctx, tx, id); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 	}
-	return stale, nil
+	return stale, mirrorIDs, nil
 }
 
 func (s *Sync) deleteMirrorSession(

internal/duckdb/sync.go

@@ -286,19 +286,13 @@ func (s *Sync) Push(
 		if err != nil {
 			return result, err
 		}
-		sessions = filterUnchangedSessions(sessions, priorFingerprints, sessionFingerprints)
-		result.Diagnostics.SkippedUnchangedSessions = skippedPushSessions(
-			candidateSessions, sessions,
-		)
 	}
-	sort.Slice(sessions, func(i, j int) bool {
-		return sessions[i].ID < sessions[j].ID
-	})
 
 	var staleIDs []string
+	var mirrorSessionIDs map[string]bool
 	err = s.withDuckTx(ctx, "delete hard-deleted sessions", func(tx *sql.Tx) error {
 		var txErr error
-		staleIDs, txErr = s.deleteHardDeletedMirrorSessions(
+		staleIDs, mirrorSessionIDs, txErr = s.deleteHardDeletedMirrorSessions(
 			ctx, tx, allLocalSessions, s.machine, s.projects, s.excludeProjects,
 		)
 		return txErr
@@ -310,6 +304,16 @@ func (s *Sync) Push(
 		delete(priorFingerprints, id)
 	}
 	result.Diagnostics.DeletedStaleSessions = len(staleIDs)
+	if !full {
+		pruneMissingMirrorFingerprints(priorFingerprints, mirrorSessionIDs)
+		sessions = filterUnchangedSessions(sessions, priorFingerprints, sessionFingerprints)
+		result.Diagnostics.SkippedUnchangedSessions = skippedPushSessions(
+			candidateSessions, sessions,
+		)
+	}
+	sort.Slice(sessions, func(i, j int) bool {
+		return sessions[i].ID < sessions[j].ID
+	})
 
 	pushed := make([]db.Session, 0, len(sessions))
 	for start := 0; start < len(sessions); start += duckSessionPushBatchSize {
@@ -877,6 +881,17 @@ func filterUnchangedSessions(
 	return out
 }
 
+func pruneMissingMirrorFingerprints(
+	priorFingerprints map[string]string,
+	mirrorSessionIDs map[string]bool,
+) {
+	for id := range priorFingerprints {
+		if !mirrorSessionIDs[id] {
+			delete(priorFingerprints, id)
+		}
+	}
+}
+
 func previousLocalSyncTimestamp(value string) (string, error) {
 	if value == "" {
 		return "", nil

internal/duckdb/sync_fastpath_test.go

@@ -352,6 +352,46 @@ func TestSyncCheckpointFailureDoesNotAdvanceWatermark(t *testing.T) {
 	assert.Empty(t, watermark)
 }
 
+func TestSyncCheckpointFailureAfterHardDeleteDoesNotKeepStaleFingerprint(t *testing.T) {
+	ctx := context.Background()
+	local := newLocalDB(t)
+	fixture := seedDuckDBSyncFixture(t, local)
+	syncer := newInMemoryTestSync(t, local, SyncOptions{})
+
+	first, err := syncer.Push(ctx, true, nil)
+	require.NoError(t, err)
+	require.Equal(t, 2, first.SessionsPushed)
+	firstWatermark, err := local.GetSyncState(lastPushStateKey)
+	require.NoError(t, err)
+	fingerprints, err := readSyncFingerprintsWithKey(local, lastPushBoundaryStateKey)
+	require.NoError(t, err)
+	require.Contains(t, fingerprints, fixture.betaID)
+
+	require.NoError(t, local.SoftDeleteSession(fixture.betaID))
+	deleted, err := local.DeleteSessionIfTrashed(fixture.betaID)
+	require.NoError(t, err)
+	require.EqualValues(t, 1, deleted)
+	policy := &checkpointSpy{err: errors.New("checkpoint failed")}
+	syncer.maintenance = policy
+
+	_, err = syncer.Push(ctx, false, nil)
+	require.ErrorContains(t, err, "checkpoint failed")
+	assert.Equal(t, 1, policy.calls)
+	assertDuckDBCountWhere(t, syncer.DB(), "sessions", "id = ?", fixture.betaID, 0)
+	watermarkAfterFailure, err := local.GetSyncState(lastPushStateKey)
+	require.NoError(t, err)
+	assert.Equal(t, firstWatermark, watermarkAfterFailure)
+
+	policy.err = nil
+	retry, err := syncer.Push(ctx, false, nil)
+	require.NoError(t, err)
+	assert.Zero(t, retry.SessionsPushed)
+	assert.Equal(t, 1, policy.calls, "retry should only repair local sync state")
+	fingerprints, err = readSyncFingerprintsWithKey(local, lastPushBoundaryStateKey)
+	require.NoError(t, err)
+	assert.NotContains(t, fingerprints, fixture.betaID)
+}
+
 func TestSyncCheckpointPolicySkipsRemoteQuackTargets(t *testing.T) {
 	ctx := context.Background()
 	local := newLocalDB(t)