Skip to content

Commit bf76433

Browse files
wesmwesm
committed
fix(data): fail closed during identity recovery
Unavailable-source recovery is the last chance to retain a resolved project when its checkout cannot be reprobed. Treating a failed snapshot lookup as resolved allowed the parser fallback to overwrite both the session and its durable evidence, so affected writes must remain retryable instead. PostgreSQL identity rows can also predate the new publication-ownership tables. On the first v3 filtered publication, adopt only ownerless rows matching that scope's successful v2 cursor so full reconciliation can remove stale metadata without claiming rows from unrelated filters.
1 parent 1cbb0bd commit bf76433

6 files changed

Lines changed: 285 additions & 15 deletions

File tree

‎internal/postgres/project_identity_upsert.go‎

Lines changed: 80 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,12 +84,21 @@ func prepareFilteredProjectIdentityPublication(
8484
ctx context.Context,
8585
q pgProjectIdentityExecer,
8686
archiveID, databaseGeneration, publicationScope string,
87-
full bool,
87+
full, adoptLegacyScope bool,
88+
projects, excludeProjects []string,
8889
observationKeys []db.ProjectIdentityObservationKey,
8990
snapshotKeys []db.SessionProjectIdentitySnapshotKey,
9091
refreshSessionIDs []string,
9192
) error {
9293
if full {
94+
if adoptLegacyScope {
95+
if err := adoptLegacyFilteredProjectIdentityScope(
96+
ctx, q, archiveID, publicationScope,
97+
projects, excludeProjects,
98+
); err != nil {
99+
return err
100+
}
101+
}
93102
if err := releaseFilteredProjectIdentityFullOwnership(
94103
ctx, q, archiveID, publicationScope,
95104
); err != nil {
@@ -111,6 +120,76 @@ func prepareFilteredProjectIdentityPublication(
111120
return nil
112121
}
113122

123+
// adoptLegacyFilteredProjectIdentityScope assigns ownerless rows written by
124+
// the v2 publisher to the filter that previously managed them. The subsequent
125+
// full reconciliation removes stale rows before publishing the current scope;
126+
// a successful v3 cursor write prevents this bounded adoption from recurring.
127+
func adoptLegacyFilteredProjectIdentityScope(
128+
ctx context.Context,
129+
q pgProjectIdentityExecer,
130+
archiveID, publicationScope string,
131+
projects, excludeProjects []string,
132+
) error {
133+
args := []any{archiveID, publicationScope}
134+
values := projects
135+
operator := "IN"
136+
if len(values) == 0 {
137+
values = excludeProjects
138+
operator = "NOT IN"
139+
}
140+
placeholders := make([]string, 0, len(values))
141+
for _, project := range values {
142+
args = append(args, project)
143+
placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
144+
}
145+
projectPredicate := operator + " (" + strings.Join(placeholders, ", ") + ")"
146+
if _, err := q.ExecContext(ctx, `
147+
INSERT INTO source_project_identity_observation_scopes (
148+
source_archive_id, project, machine, root_path, git_remote,
149+
publication_scope
150+
)
151+
SELECT observation.source_archive_id, observation.project,
152+
observation.machine, observation.root_path, observation.git_remote, $2
153+
FROM source_project_identity_observations observation
154+
WHERE observation.source_archive_id = $1
155+
AND observation.project `+projectPredicate+`
156+
AND NOT EXISTS (
157+
SELECT 1
158+
FROM source_project_identity_observation_scopes owner
159+
WHERE owner.source_archive_id = observation.source_archive_id
160+
AND owner.project = observation.project
161+
AND owner.machine = observation.machine
162+
AND owner.root_path = observation.root_path
163+
AND owner.git_remote = observation.git_remote
164+
)
165+
ON CONFLICT DO NOTHING`, args...); err != nil {
166+
return fmt.Errorf("adopting legacy pg identity observations: %w", err)
167+
}
168+
169+
if _, err := q.ExecContext(ctx, `
170+
INSERT INTO source_session_project_identity_snapshot_scopes (
171+
source_archive_id, source_database_generation,
172+
source_session_id, publication_scope
173+
)
174+
SELECT snapshot.source_archive_id, snapshot.source_database_generation,
175+
snapshot.source_session_id, $2
176+
FROM source_session_project_identity_snapshots snapshot
177+
WHERE snapshot.source_archive_id = $1
178+
AND snapshot.project `+projectPredicate+`
179+
AND NOT EXISTS (
180+
SELECT 1
181+
FROM source_session_project_identity_snapshot_scopes owner
182+
WHERE owner.source_archive_id = snapshot.source_archive_id
183+
AND owner.source_database_generation =
184+
snapshot.source_database_generation
185+
AND owner.source_session_id = snapshot.source_session_id
186+
)
187+
ON CONFLICT DO NOTHING`, args...); err != nil {
188+
return fmt.Errorf("adopting legacy pg identity snapshots: %w", err)
189+
}
190+
return nil
191+
}
192+
114193
func releaseFilteredProjectIdentityFullOwnership(
115194
ctx context.Context,
116195
q pgProjectIdentityExecer,

‎internal/postgres/push.go‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ const (
2626
lastPushSourceArchiveIDKey = "pg_source_archive_id_v1"
2727
lastPushTargetFingerprintKey = "pg_target_fingerprint_v1"
2828
sessionAliasBackfillStateKey = "pg_session_alias_backfill_v1"
29+
legacyProjectIdentityStateKey = "project_identity_publication_revision_v2"
2930
projectIdentityPublicationStateKey = "project_identity_publication_revision_v3"
3031
transcriptRevisionBackfillStateKey = "pg_transcript_revision_backfill_v1"
3132
sessionProvenanceBackfillStateKey = "pg_session_provenance_backfill_v1"
@@ -763,6 +764,19 @@ func (s *Sync) syncProjectIdentityObservations(
763764
if err != nil {
764765
return fmt.Errorf("reading project identity publication revision: %w", err)
765766
}
767+
adoptLegacyFilteredScope := false
768+
if s.isFiltered() && publishedRevisionValue == "" {
769+
legacyValue, loadErr := state.GetSyncState(
770+
legacyProjectIdentityStateKey + ":" + databaseGeneration,
771+
)
772+
if loadErr != nil {
773+
return fmt.Errorf(
774+
"reading legacy project identity publication revision: %w",
775+
loadErr,
776+
)
777+
}
778+
adoptLegacyFilteredScope = legacyValue != ""
779+
}
766780
fullPublication := force || publishedRevisionValue == ""
767781
var publishedRevision int64
768782
if !fullPublication {
@@ -846,8 +860,9 @@ func (s *Sync) syncProjectIdentityObservations(
846860
)
847861
if err := prepareFilteredProjectIdentityPublication(
848862
ctx, tx, archiveID, databaseGeneration, publicationScope,
849-
fullPublication, delta.ObservationDeletes, delta.SnapshotDeletes,
850-
refreshSessionIDs,
863+
fullPublication, adoptLegacyFilteredScope,
864+
s.projects, s.excludeProjects,
865+
delta.ObservationDeletes, delta.SnapshotDeletes, refreshSessionIDs,
851866
); err != nil {
852867
return err
853868
}

‎internal/postgres/push_pgtest_test.go‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,102 @@ func TestFilteredIdentityPublicationPreservesUnfilteredMetadata(t *testing.T) {
238238
}
239239
}
240240

241+
func TestFilteredIdentityPublicationAdoptsLegacyOwnerlessScope(t *testing.T) {
242+
const (
243+
schema = "agentsview_legacy_identity_scope_test"
244+
includedProject = "included_project"
245+
excludedProject = "excluded_project"
246+
)
247+
pgURL := testPGURL(t)
248+
cleanNamedPGSchema(t, pgURL, schema)
249+
t.Cleanup(func() { cleanNamedPGSchema(t, pgURL, schema) })
250+
ctx := context.Background()
251+
local, err := db.Open(filepath.Join(t.TempDir(), "local.db"))
252+
require.NoError(t, err)
253+
t.Cleanup(func() { require.NoError(t, local.Close()) })
254+
for _, fixture := range []struct {
255+
project string
256+
root string
257+
}{
258+
{includedProject, "/workspace/included"},
259+
{excludedProject, "/workspace/excluded"},
260+
} {
261+
sessionID := "identity-" + fixture.project
262+
require.NoError(t, local.UpsertSession(db.Session{
263+
ID: sessionID, Project: fixture.project,
264+
Machine: "test-machine", Agent: "codex", Cwd: fixture.root,
265+
}))
266+
require.NoError(t, local.UpsertProjectIdentityObservation(
267+
ctx, export.ProjectIdentityObservation{
268+
SessionID: sessionID, Project: fixture.project,
269+
Machine: "test-machine", RootPath: fixture.root,
270+
GitRemote: "https://example.com/team/" + fixture.project + ".git",
271+
RemoteResolution: export.ProjectResolutionResolved,
272+
ObservedAt: time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC),
273+
},
274+
))
275+
}
276+
277+
unfiltered, err := New(
278+
pgURL, schema, local, "test-machine", true, SyncOptions{},
279+
)
280+
require.NoError(t, err)
281+
require.NoError(t, unfiltered.EnsureSchema(ctx))
282+
require.NoError(t, unfiltered.syncProjectIdentityObservations(ctx, false, nil))
283+
require.NoError(t, unfiltered.Close())
284+
285+
pg, err := Open(pgURL, schema, true)
286+
require.NoError(t, err)
287+
_, err = pg.ExecContext(ctx, `
288+
DELETE FROM source_project_identity_observation_scopes;
289+
DELETE FROM source_session_project_identity_snapshot_scopes`)
290+
require.NoError(t, err)
291+
require.NoError(t, pg.Close())
292+
293+
filtered, err := New(
294+
pgURL, schema, local, "test-machine", true,
295+
SyncOptions{Projects: []string{includedProject}},
296+
)
297+
require.NoError(t, err)
298+
t.Cleanup(func() { require.NoError(t, filtered.Close()) })
299+
require.NoError(t, filtered.EnsureSchema(ctx))
300+
generation, err := local.GetDatabaseID(ctx)
301+
require.NoError(t, err)
302+
require.NoError(t, filtered.effectiveSyncState().SetSyncState(
303+
legacyProjectIdentityStateKey+":"+generation, "1",
304+
))
305+
_, err = local.CreateWorktreeProjectMapping(ctx, db.WorktreeProjectMapping{
306+
Machine: "test-machine", PathPrefix: "/workspace/included",
307+
Layout: db.WorktreeMappingLayoutExplicit, Project: excludedProject,
308+
OriginalProject: includedProject, Enabled: true,
309+
})
310+
require.NoError(t, err)
311+
applied, err := local.ApplyWorktreeProjectMappings(ctx, "test-machine")
312+
require.NoError(t, err)
313+
require.Equal(t, 1, applied.UpdatedSessions)
314+
315+
require.NoError(t, filtered.syncProjectIdentityObservations(ctx, false, nil))
316+
for _, table := range []string{
317+
"source_project_identity_observations",
318+
"source_session_project_identity_snapshots",
319+
} {
320+
rows, queryErr := filtered.pg.QueryContext(ctx,
321+
"SELECT project FROM "+table+" ORDER BY project",
322+
)
323+
require.NoError(t, queryErr)
324+
var projects []string
325+
for rows.Next() {
326+
var project string
327+
require.NoError(t, rows.Scan(&project))
328+
projects = append(projects, project)
329+
}
330+
require.NoError(t, rows.Err())
331+
require.NoError(t, rows.Close())
332+
assert.Equal(t, []string{excludedProject}, projects,
333+
"first v3 publication must remove stale rows from its legacy scope")
334+
}
335+
}
336+
241337
func TestFilteredThenUnfilteredIdentityPublicationIncludesExcludedProject(
242338
t *testing.T,
243339
) {

‎internal/sync/engine.go‎

Lines changed: 35 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -10069,23 +10069,25 @@ func (e *Engine) loadWorktreeProjectResolver() worktreeProjectResolver {
1006910069
}
1007010070

1007110071
func (e *Engine) preserveUnavailableSourceProjects(
10072+
ctx context.Context,
1007210073
batch []pendingWrite,
10073-
) []pendingWrite {
10074+
) ([]pendingWrite, error) {
1007410075
indexes := make(map[string][]int)
1007510076
ids := make([]string, 0, len(batch))
1007610077
for i := range batch {
1007710078
if batch[i].sourceProjectResolved {
1007810079
continue
1007910080
}
10080-
batch[i].sourceProjectResolved = true
1008110081
sess := batch[i].sess
1008210082
if sess.ID == "" || sess.Project == "" || sess.Cwd == "" ||
1008310083
sess.Machine != e.machine ||
1008410084
!safeLocalAbsolutePath(sess.Cwd) ||
1008510085
export.IsAutomountNamespacePath(runtime.GOOS, filepath.Clean(sess.Cwd)) {
10086+
batch[i].sourceProjectResolved = true
1008610087
continue
1008710088
}
1008810089
if _, err := os.Stat(sess.Cwd); !errors.Is(err, os.ErrNotExist) {
10090+
batch[i].sourceProjectResolved = true
1008910091
continue
1009010092
}
1009110093
if _, exists := indexes[sess.ID]; !exists {
@@ -10094,17 +10096,21 @@ func (e *Engine) preserveUnavailableSourceProjects(
1009410096
indexes[sess.ID] = append(indexes[sess.ID], i)
1009510097
}
1009610098
if len(ids) == 0 {
10097-
return batch
10099+
return batch, nil
1009810100
}
1009910101

1010010102
snapshots, err := e.db.ListSessionProjectIdentitySnapshotsByID(
10101-
context.Background(), ids,
10103+
ctx, ids,
1010210104
)
1010310105
if err != nil {
10104-
log.Printf(
10105-
"load unavailable-cwd project identity snapshots: %v", err,
10106+
return batch, fmt.Errorf(
10107+
"load unavailable-cwd project identity snapshots: %w", err,
1010610108
)
10107-
return batch
10109+
}
10110+
for _, matchingIndexes := range indexes {
10111+
for _, i := range matchingIndexes {
10112+
batch[i].sourceProjectResolved = true
10113+
}
1010810114
}
1010910115
for id, snapshot := range snapshots {
1011010116
if snapshot.Project == "" ||
@@ -10121,7 +10127,7 @@ func (e *Engine) preserveUnavailableSourceProjects(
1012110127
sess.Project = snapshot.Project
1012210128
}
1012310129
}
10124-
return batch
10130+
return batch, nil
1012510131
}
1012610132

1012710133
func pathContains(root, path string) bool {
@@ -10170,7 +10176,19 @@ func (e *Engine) writeBatchWithOutcome(
1017010176
writeMode syncWriteMode,
1017110177
forceReplace bool,
1017210178
) writeBatchOutcome {
10173-
batch = e.preserveUnavailableSourceProjects(batch)
10179+
var err error
10180+
batch, err = e.preserveUnavailableSourceProjects(
10181+
context.Background(), batch,
10182+
)
10183+
if err != nil {
10184+
log.Printf("preserve unavailable source projects: %v", err)
10185+
outcome := writeBatchOutcome{written: make([]bool, len(batch))}
10186+
for _, pw := range batch {
10187+
e.markStaleFailedMemberWrite(pw)
10188+
}
10189+
outcome.failedSessions = len(batch)
10190+
return outcome
10191+
}
1017410192
if writeMode == syncWriteBulk {
1017510193
return e.writeBatchBulkWithOutcome(batch, forceReplace)
1017610194
}
@@ -10315,7 +10333,6 @@ func (e *Engine) prepareSessionWrite(
1031510333
pw pendingWrite,
1031610334
resolveWorktreeProject worktreeProjectResolver,
1031710335
) (db.Session, []db.Message, sessionWriteVerdict) {
10318-
pw = e.preserveUnavailableSourceProjects([]pendingWrite{pw})[0]
1031910336
msgs := toDBMessages(pw, e.blockedResultCategories)
1032010337
s := toDBSession(pw)
1032110338
applySessionMessageDerivedFields(
@@ -11851,14 +11868,20 @@ func (e *Engine) writeSessionFullWithResolver(
1185111868
pw pendingWrite,
1185211869
resolveWorktreeProject worktreeProjectResolver,
1185311870
) error {
11854-
pw = e.preserveUnavailableSourceProjects([]pendingWrite{pw})[0]
11871+
preserved, err := e.preserveUnavailableSourceProjects(
11872+
context.Background(), []pendingWrite{pw},
11873+
)
11874+
if err != nil {
11875+
return err
11876+
}
11877+
pw = preserved[0]
1185511878
s, msgs, verdict := e.prepareSessionWrite(
1185611879
pw, resolveWorktreeProject,
1185711880
)
1185811881
if verdict != sessionWriteOK {
1185911882
return errSessionPreserved
1186011883
}
11861-
_, err := e.upsertSessionPendingContentWithProjectIdentity(
11884+
_, err = e.upsertSessionPendingContentWithProjectIdentity(
1186211885
s, pw.sess.Project,
1186311886
)
1186411887
if err != nil {

0 commit comments

Comments
 (0)