fix(postgres): scope generation witnesses to each watcher

rodboev · rodboev · commit f1dc6daed7d5 · 2026-07-23T17:29:28.000-04:00
diff --git a/internal/postgres/vector_admin.go b/internal/postgres/vector_admin.go
@@ -129,14 +129,25 @@ func vectorGenerationMachines(
 		if err := rows.Scan(&m); err != nil {
 			return nil, fmt.Errorf("scanning generation machine: %w", err)
 		}
-		machines = append(machines, m)
+		display := vectorGenerationMachineDisplayName(m)
+		if len(machines) > 0 && machines[len(machines)-1] == display {
+			continue
+		}
+		machines = append(machines, display)
 	}
 	if err := rows.Err(); err != nil {
 		return nil, fmt.Errorf("iterating generation machines: %w", err)
 	}
 	return machines, nil
 }
 
+func vectorGenerationMachineDisplayName(raw string) string {
+	if i := strings.Index(raw, "|"+pushMarkerKeyPrefix); i >= 0 {
+		return raw[:i]
+	}
+	return raw
+}
+
 // DropVectorGeneration removes a generation and all of its data: its chunk
 // table, its vector_push_state and vector_generation_machines rows, and its
 // vector_generations row, then prunes vector_documents rows referenced by no
diff --git a/internal/postgres/vector_push.go b/internal/postgres/vector_push.go
@@ -123,11 +123,12 @@ const vectorProgressStride = 2000
 // the schema-qualified halfvec type. The type must be qualified because the
 // connection's search_path is the target schema only, while pgvector's types
 // live in whichever schema first installed the extension.
-// machineRecorded reports whether this machine's push record already existed
-// for the generation before this push touched it: recreating the vector
-// tables restarts the id sequence, so a reused id can satisfy the memo
-// comparison while the recreated generation is empty, and the machine record
-// — wiped in the same reset — is the witness that survives id reuse.
+// machineRecorded reports whether this pusher's scoped witness record already
+// existed for the generation before this push touched it: recreating the
+// vector tables restarts the id sequence, so a reused id can satisfy the memo
+// comparison while the recreated generation is empty, and the witness record
+// — keyed by local push marker plus sync/filter scope and wiped in the same
+// reset — is what survives id reuse safely.
 type vectorGeneration struct {
 	id              int64
 	halfvecType     string
@@ -270,26 +271,29 @@ func (s *Sync) pushVectors(
 		res.Skipped, res.SkippedReason = true, unavailable
 		return res, nil
 	}
+	witnessKey, err := s.vectorGenerationWitnessKey()
+	if err != nil {
+		return res, err
+	}
 	// A scoped push writes only its changed sessions' chunks, which is safe
 	// only within the exact generation instance this process last reconciled
 	// generation-wide. Two signals identify that instance. The PG generation
 	// id catches every recreation that keeps the tables: row deletion never
 	// resets the sequence, so a re-embed (new fingerprint), reset, or admin
 	// drop yields a new id, whoever recreates it. Recreating the tables
 	// themselves restarts the sequence, so a reused id can match a stale
-	// memo; this machine's push record, wiped in the same reset and written
-	// only after a clean generation-wide reconciliation, is the witness for
-	// that case — a scoped push always follows a generation-wide push in the
-	// same process, which recorded it. Either signal failing means the prior
-	// reconciliation no longer covers this generation and scoping would
-	// leave search reading an incomplete one until the interval floor.
+	// memo; this pusher's scoped witness record, wiped in the same reset and
+	// written only after a clean generation-wide reconciliation, is the
+	// witness for that case. Either signal failing means the prior
+	// reconciliation no longer covers this generation and scoping would leave
+	// search reading an incomplete one until the interval floor.
 	// Promote to a generation-wide read. A zero memo means no reconciliation
 	// to trust yet, so the reconcile bit already forces this push
 	// generation-wide and the id check must not fire.
 	requestedScope := scope
 	var resolved vectorGeneration
 	if scope != nil {
-		initialProbe, found, err := s.lookupVectorGeneration(ctx, gen.Fingerprint)
+		initialProbe, found, err := s.lookupVectorGeneration(ctx, gen.Fingerprint, witnessKey)
 		if err != nil {
 			return res, err
 		}
@@ -298,8 +302,8 @@ func (s *Sync) pushVectors(
 				gen.Fingerprint)
 			scope = nil
 		} else if !initialProbe.machineRecorded {
-			log.Printf("vector push: no prior push record for machine %q against generation %d; promoting scoped push to generation-wide reconciliation",
-				s.machine, initialProbe.id)
+			log.Printf("vector push: no prior scoped witness against generation %d; promoting scoped push to generation-wide reconciliation",
+				initialProbe.id)
 			scope = nil
 		} else if lastReconciledGeneration != 0 &&
 			initialProbe.id != lastReconciledGeneration {
@@ -312,7 +316,7 @@ func (s *Sync) pushVectors(
 				s.afterVectorGenerationLookup = nil
 				hook()
 			}
-			currentProbe, found, err := s.lookupVectorGeneration(ctx, gen.Fingerprint)
+			currentProbe, found, err := s.lookupVectorGeneration(ctx, gen.Fingerprint, witnessKey)
 			if err != nil {
 				return res, err
 			}
@@ -321,8 +325,8 @@ func (s *Sync) pushVectors(
 					gen.Fingerprint)
 				scope = nil
 			} else if !currentProbe.machineRecorded {
-				log.Printf("vector push: no prior push record for machine %q against generation %d before scoped reconciliation; promoting scoped push to generation-wide reconciliation",
-					s.machine, currentProbe.id)
+				log.Printf("vector push: no prior scoped witness against generation %d before scoped reconciliation; promoting scoped push to generation-wide reconciliation",
+					currentProbe.id)
 				scope = nil
 			} else if currentProbe.id != initialProbe.id {
 				log.Printf("vector push: active generation id changed from %d to %d before scoped reconciliation; promoting scoped push to generation-wide reconciliation",
@@ -358,7 +362,7 @@ func (s *Sync) pushVectors(
 		gen = export.Generation()
 	}
 	if scope == nil {
-		resolved, err = s.resolveVectorGeneration(ctx, gen)
+		resolved, err = s.resolveVectorGeneration(ctx, gen, witnessKey)
 		if err != nil {
 			if s.skipVectorsOnPrivilegeError(err, &res) {
 				return res, nil
@@ -413,7 +417,7 @@ func (s *Sync) pushVectors(
 				ctx, full, nil, lastReconciledGeneration, failedSessions, onProgress,
 			)
 		}
-		finalProbe, found, err := s.lookupVectorGeneration(ctx, gen.Fingerprint)
+		finalProbe, found, err := s.lookupVectorGeneration(ctx, gen.Fingerprint, witnessKey)
 		if err != nil {
 			return res, err
 		}
@@ -425,8 +429,8 @@ func (s *Sync) pushVectors(
 		}
 		if !finalProbe.machineRecorded {
 			return retryGenerationWide(
-				"vector push: no prior push record for machine %q against generation %d after scoped reconciliation; retrying generation-wide",
-				s.machine, finalProbe.id,
+				"vector push: no prior scoped witness against generation %d after scoped reconciliation; retrying generation-wide",
+				finalProbe.id,
 			)
 		}
 		if finalProbe.id != resolved.id {
@@ -437,7 +441,7 @@ func (s *Sync) pushVectors(
 		}
 	}
 	if scope == nil && res.SessionsDeferred == 0 {
-		if err := s.recordVectorGenerationMachine(ctx, resolved.id); err != nil {
+		if err := s.recordVectorGenerationMachine(ctx, resolved.id, witnessKey); err != nil {
 			return res, err
 		}
 	}
@@ -448,7 +452,7 @@ func (s *Sync) pushVectors(
 }
 
 func (s *Sync) lookupVectorGeneration(
-	ctx context.Context, fingerprint string,
+	ctx context.Context, fingerprint, witnessKey string,
 ) (vectorGeneration, bool, error) {
 	genID, _, ok, err := LookupVectorGeneration(ctx, s.pg, fingerprint)
 	if err != nil {
@@ -466,7 +470,7 @@ func (s *Sync) lookupVectorGeneration(
 SELECT EXISTS (
     SELECT 1 FROM vector_generation_machines
      WHERE generation_id = $1 AND machine = $2)`,
-		genID, s.machine).Scan(&machineRecorded); err != nil {
+		genID, witnessKey).Scan(&machineRecorded); err != nil {
 		return vectorGeneration{}, false, fmt.Errorf(
 			"reading vector push machine record: %w", err,
 		)
@@ -801,11 +805,11 @@ func vectorOutOfScopeQuery(
 // against it. Its id is the generation
 // instance's identity, which a scoped push compares against the last one it
 // reconciled generation-wide to decide whether to promote; whether this
-// machine's push record predated this call is captured first, because that
-// record is the incarnation witness the promotion check falls back on when a
-// recreated id sequence hands the new generation the memoized id.
+// scoped witness predated this call is captured first, because that record is
+// the incarnation witness the promotion check falls back on when a recreated
+// id sequence hands the new generation the memoized id.
 func (s *Sync) resolveVectorGeneration(
-	ctx context.Context, gen VectorGenerationInfo,
+	ctx context.Context, gen VectorGenerationInfo, witnessKey string,
 ) (vectorGeneration, error) {
 	genID, err := ensureVectorGeneration(
 		ctx, s.pg, gen.Fingerprint, gen.Model, gen.Dimension,
@@ -825,7 +829,7 @@ func (s *Sync) resolveVectorGeneration(
 SELECT EXISTS (
     SELECT 1 FROM vector_generation_machines
      WHERE generation_id = $1 AND machine = $2)`,
-		genID, s.machine).Scan(&machineRecorded); err != nil {
+		genID, witnessKey).Scan(&machineRecorded); err != nil {
 		return vectorGeneration{}, fmt.Errorf("reading vector push machine record: %w", err)
 	}
 	return vectorGeneration{
@@ -835,14 +839,22 @@ SELECT EXISTS (
 	}, nil
 }
 
+func (s *Sync) vectorGenerationWitnessKey() (string, error) {
+	markerID, err := s.pushMarkerID()
+	if err != nil {
+		return "", err
+	}
+	return s.machine + "|" + s.pushMarkerMetadataKey(pushMarkerKeyPrefix, markerID), nil
+}
+
 func (s *Sync) recordVectorGenerationMachine(
-	ctx context.Context, genID int64,
+	ctx context.Context, genID int64, witnessKey string,
 ) error {
 	if _, err := s.pg.ExecContext(ctx, `
 INSERT INTO vector_generation_machines (generation_id, machine, last_push_at)
 VALUES ($1, $2, now())
 ON CONFLICT (generation_id, machine) DO UPDATE SET last_push_at = now()`,
-		genID, s.machine); err != nil {
+		genID, witnessKey); err != nil {
 		return fmt.Errorf("recording vector push machine: %w", err)
 	}
 	return nil
diff --git a/internal/postgres/vector_push_pg_test.go b/internal/postgres/vector_push_pg_test.go
@@ -787,6 +787,8 @@ func TestVectorPushDeferredFullPassDoesNotRecordMachineWitness(t *testing.T) {
 	require.NoError(t, err, "baseline Push")
 	genID := res.Vectors.GenerationID
 	require.NotZero(t, genID)
+	witnessKey, err := sync.vectorGenerationWitnessKey()
+	require.NoError(t, err, "vectorGenerationWitnessKey")
 	_, err = pg.Exec(
 		`DELETE FROM vector_generation_machines WHERE generation_id = $1`, genID,
 	)
@@ -805,7 +807,7 @@ func TestVectorPushDeferredFullPassDoesNotRecordMachineWitness(t *testing.T) {
 	assert.Equal(t, 1, vres.SessionsPushed)
 	assert.Equal(t, 0, countRows(t, pg, `
 SELECT COUNT(*) FROM vector_generation_machines
- WHERE generation_id = $1 AND machine = $2`, genID, "test-machine"),
+ WHERE generation_id = $1 AND machine = $2`, genID, witnessKey),
 		"deferred generation-wide work must not witness the generation")
 
 	src.genScopes = nil
@@ -823,14 +825,109 @@ SELECT COUNT(*) FROM vector_generation_machines
 		"the promoted retry repairs B's deferred vector state")
 	assert.Equal(t, 1, countRows(t, pg, `
 SELECT COUNT(*) FROM vector_generation_machines
- WHERE generation_id = $1 AND machine = $2`, genID, "test-machine"),
+ WHERE generation_id = $1 AND machine = $2`, genID, witnessKey),
 		"the clean generation-wide retry records the witness")
 	assert.Equal(t, 1, countRows(t, pg, `
 SELECT COUNT(*) FROM vector_push_state
  WHERE generation_id = $1 AND session_id = 'B' AND doc_agg_hash = 'b2'`, genID),
 		"the promoted retry must repair the deferred out-of-scope session")
 }
 
+// TestVectorPushFilteredWitnessDoesNotCrossScopesAfterTableRecreation pins the
+// filtered-watcher incarnation bug: after vector tables are recreated on the
+// same machine, one filter's generation-wide witness must not let another
+// filter trust a reused generation id for a scoped push.
+func TestVectorPushFilteredWitnessDoesNotCrossScopesAfterTableRecreation(t *testing.T) {
+	pgURL := testPGURL(t)
+	_, localDB, pg := newVectorPushTestSync(
+		t, pgURL, "agentsview_vector_push_filtered_witness_test")
+	ctx := context.Background()
+
+	seedVectorSessionProject(t, localDB, "alpha", "alpha")
+	seedVectorSessionProject(t, localDB, "beta-1", "beta")
+	seedVectorSessionProject(t, localDB, "beta-2", "beta")
+
+	src := &fakeVectorSource{
+		gen:    VectorGenerationInfo{Fingerprint: "fp-filtered", Model: "m", Dimension: 4},
+		hasGen: true,
+		hashes: map[string]string{"alpha": "ha", "beta-1": "hb1", "beta-2": "hb2"},
+		docs: map[string][]VectorPushDoc{
+			"alpha":  {vdoc("alpha", "alpha#0", 0, "a", "ha", []float32{1, 0, 0, 0})},
+			"beta-1": {vdoc("beta-1", "beta-1#0", 0, "b1", "hb1", []float32{0, 1, 0, 0})},
+			"beta-2": {vdoc("beta-2", "beta-2#0", 0, "b2", "hb2", []float32{0, 0, 1, 0})},
+		},
+	}
+	filterScopeAlpha := pushSyncStateScope("work", []string{"alpha"}, nil)
+	filterScopeBeta := pushSyncStateScope("work", []string{"beta"}, nil)
+	targetState := newScopedSyncStateStore(localDB, "work", false)
+	syncAlpha := &Sync{
+		pg:                 pg,
+		local:              localDB,
+		syncState:          newScopedSyncStateStore(localDB, filterScopeAlpha, false),
+		aliasBackfillState: targetState,
+		machine:            "test-machine",
+		schema:             "agentsview_vector_push_filtered_witness_test",
+		targetFingerprint:  "target-fp",
+		syncStateTarget:    filterScopeAlpha,
+		projects:           []string{"alpha"},
+		vectorSource:       src,
+		schemaDone:         true,
+	}
+	syncBeta := &Sync{
+		pg:                 pg,
+		local:              localDB,
+		syncState:          newScopedSyncStateStore(localDB, filterScopeBeta, false),
+		aliasBackfillState: targetState,
+		machine:            "test-machine",
+		schema:             "agentsview_vector_push_filtered_witness_test",
+		targetFingerprint:  "target-fp",
+		syncStateTarget:    filterScopeBeta,
+		projects:           []string{"beta"},
+		vectorSource:       src,
+		schemaDone:         true,
+	}
+
+	base, err := syncBeta.Push(ctx, false, nil)
+	require.NoError(t, err, "baseline beta Push")
+	gen1 := base.Vectors.GenerationID
+	require.NotZero(t, gen1)
+
+	for _, q := range []string{
+		`DROP TABLE IF EXISTS ` + vectorChunkTable(gen1),
+		`DROP TABLE IF EXISTS vector_push_state`,
+		`DROP TABLE IF EXISTS vector_generation_machines`,
+		`DROP TABLE IF EXISTS vector_documents`,
+		`DROP TABLE IF EXISTS vector_generations`,
+	} {
+		_, err := pg.Exec(q)
+		require.NoError(t, err, q)
+	}
+
+	alphaRes, err := syncAlpha.Push(ctx, false, nil)
+	require.NoError(t, err, "alpha Push after table recreation")
+	require.Equal(t, gen1, alphaRes.Vectors.GenerationID,
+		"table recreation reuses the generation id in the case under test")
+
+	src.genScopes = nil
+	src.hashScopes = nil
+	vres, err := syncBeta.pushVectors(ctx, false, []string{"beta-1"}, gen1, nil, nil)
+	require.NoError(t, err, "beta scoped push after alpha witness replay")
+	assert.Equal(t, 2, vres.SessionsPushed,
+		"beta scope must repopulate every in-scope session generation-wide")
+	require.Len(t, src.genScopes, 2)
+	assert.Equal(t, []string{"beta-1"}, src.genScopes[0],
+		"the first pass starts scoped")
+	assert.Nil(t, src.genScopes[1],
+		"the beta watcher must reopen generation-wide instead of trusting alpha's witness")
+	require.Len(t, src.hashScopes, 1)
+	assert.Nil(t, src.hashScopes[0],
+		"the replay must read every beta session, not only beta-1")
+	assert.Equal(t, 2, countRows(t, pg, `
+SELECT COUNT(*) FROM vector_push_state
+ WHERE generation_id = $1 AND session_id IN ('beta-1', 'beta-2')`, gen1),
+		"both beta sessions have state rows under the recreated generation")
+}
+
 func TestVectorPushRoundTrip(t *testing.T) {
 	pgURL := testPGURL(t)
 	sync, localDB, pg := newVectorPushTestSync(
@@ -884,6 +981,8 @@ func TestVectorPushRoundTrip(t *testing.T) {
 	require.NoError(t, err, "LookupVectorGeneration")
 	require.True(t, ok, "generation registered")
 	assert.Equal(t, 4, dim)
+	witnessKey, err := sync.vectorGenerationWitnessKey()
+	require.NoError(t, err, "vectorGenerationWitnessKey")
 
 	assert.Equal(t, 3,
 		countRows(t, pg, `SELECT COUNT(*) FROM vector_documents`))
@@ -893,7 +992,7 @@ func TestVectorPushRoundTrip(t *testing.T) {
 		`SELECT COUNT(*) FROM vector_push_state WHERE generation_id = $1`, genID))
 	assert.Equal(t, 1, countRows(t, pg,
 		`SELECT COUNT(*) FROM vector_generation_machines
-		 WHERE generation_id = $1 AND machine = $2`, genID, "test-machine"))
+		 WHERE generation_id = $1 AND machine = $2`, genID, witnessKey))
 }
 
 func TestVectorPushDeltaNoop(t *testing.T) {
