Skip to content

Commit ad7ae9e

Browse files
wesmwesm
committed
fix(sync): revalidate the container capture before watermark-only filtering
The changed-path filter compared watermark-only listings against stored state with no proof the container was stable across the listing window: a commit landing between the watermark query and filtering could advance a session past its listed watermark, and when every source filtered out the pass ended before beginSQLiteContainerPass ever ran its recapture check, so nothing caught the skew until the next write. Classification now captures the container's state before any watermark-only listing and the filter compares a fresh recapture against it, keeping every source for full fingerprinting when the capture is missing or stale.
1 parent 1bdad32 commit ad7ae9e

3 files changed

Lines changed: 145 additions & 5 deletions

File tree

‎internal/sync/engine.go‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1081,6 +1081,20 @@ func (e *Engine) classifyProviderChangedPath(
10811081
!changedPathWithinAnyRoot(path, watchRoots) {
10821082
continue
10831083
}
1084+
// Capture the shared container's state before any watermark-only
1085+
// listing below: filterFreshWatermarkOnlySources may trust such a
1086+
// listing only while the container provably has not changed since
1087+
// this capture, and the pass-level capture guard does not exist yet
1088+
// at classification time.
1089+
watermarkContainer := openCodeContainerPathForChangedPathEvent(
1090+
agentType, roots, path,
1091+
)
1092+
var watermarkPreState parser.SQLiteContainerState
1093+
watermarkPreStateOK := false
1094+
if watermarkContainer != "" {
1095+
watermarkPreState, watermarkPreStateOK =
1096+
statSQLiteContainerState(watermarkContainer)
1097+
}
10841098
for _, watchRoot := range watchRoots {
10851099
request := parser.ChangedPathRequest{
10861100
Path: path,
@@ -1133,7 +1147,8 @@ func (e *Engine) classifyProviderChangedPath(
11331147
continue
11341148
}
11351149
sources = e.filterFreshWatermarkOnlySources(
1136-
ctx, agentType, roots, path, sources,
1150+
ctx, watermarkContainer, watermarkPreState,
1151+
watermarkPreStateOK, sources,
11371152
)
11381153
if def.Type == parser.AgentOmnigent {
11391154
sources, err = e.expandOmnigentInheritedMetadataSources(

‎internal/sync/opencode_container_gate.go‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,14 @@ func openCodeContainerPathForChangedPathEvent(
174174
// — and sessions with no stored row or a stale data version are kept
175175
// unconditionally.
176176
//
177+
// The comparison is only trustworthy while the container provably has not
178+
// changed since preState, captured before the watermark listing ran: a
179+
// commit landing inside that window can advance a session past its listed
180+
// watermark, and no pass-level capture guard exists yet at classification
181+
// time — if every source were filtered here, the pass would end before
182+
// beginSQLiteContainerPass ever ran its recapture check. A missing or
183+
// mismatched capture keeps every source for full fingerprinting.
184+
//
177185
// Known, deliberate deferral (not a detection gap to "fix" here): a
178186
// child-only write that leaves the session and project rows untouched is
179187
// invisible to the session-row watermark wherever its timestamps land —
@@ -189,15 +197,14 @@ func openCodeContainerPathForChangedPathEvent(
189197
// decides instead.
190198
func (e *Engine) filterFreshWatermarkOnlySources(
191199
ctx context.Context,
192-
agent parser.AgentType,
193-
roots []string,
194-
path string,
200+
container string,
201+
preState parser.SQLiteContainerState,
202+
preStateOK bool,
195203
sources []parser.SourceRef,
196204
) []parser.SourceRef {
197205
if len(sources) == 0 || e.forceParse || e.pathRewriter != nil {
198206
return sources
199207
}
200-
container := openCodeContainerPathForChangedPathEvent(agent, roots, path)
201208
if container == "" {
202209
return sources
203210
}
@@ -211,6 +218,13 @@ func (e *Engine) filterFreshWatermarkOnlySources(
211218
if !watermarkOnly {
212219
return sources
213220
}
221+
if !preStateOK {
222+
return sources
223+
}
224+
if post, ok := statSQLiteContainerState(container); !ok ||
225+
post != preState {
226+
return sources
227+
}
214228
stored, err := e.db.ListVirtualContainerMemberFreshness(ctx, container)
215229
if err != nil || len(stored) == 0 {
216230
return sources

‎internal/sync/opencode_container_gate_internal_test.go‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,117 @@ func newContainerTestDB(t *testing.T) (string, *sql.DB) {
2525
return path, conn
2626
}
2727

28+
// newCompositeContainerTestDB creates an OpenCode container whose schema
29+
// carries the composite change-signal columns and session_id indexes, so
30+
// watermark-only listings are supported.
31+
func newCompositeContainerTestDB(t *testing.T) (string, *sql.DB) {
32+
t.Helper()
33+
path := filepath.Join(t.TempDir(), "opencode.db")
34+
conn, err := sql.Open("sqlite3", path)
35+
require.NoError(t, err, "open container db")
36+
t.Cleanup(func() { _ = conn.Close() })
37+
_, err = conn.Exec(`
38+
CREATE TABLE project (
39+
id TEXT PRIMARY KEY,
40+
worktree TEXT NOT NULL,
41+
time_updated INTEGER NOT NULL DEFAULT 0
42+
);
43+
CREATE TABLE session (
44+
id TEXT PRIMARY KEY,
45+
project_id TEXT NOT NULL,
46+
time_created INTEGER NOT NULL,
47+
time_updated INTEGER NOT NULL
48+
);
49+
CREATE TABLE message (
50+
id TEXT PRIMARY KEY,
51+
session_id TEXT NOT NULL,
52+
data TEXT NOT NULL,
53+
time_created INTEGER NOT NULL,
54+
time_updated INTEGER NOT NULL DEFAULT 0
55+
);
56+
CREATE TABLE part (
57+
id TEXT PRIMARY KEY,
58+
session_id TEXT NOT NULL,
59+
message_id TEXT NOT NULL,
60+
data TEXT NOT NULL,
61+
time_created INTEGER NOT NULL,
62+
time_updated INTEGER NOT NULL DEFAULT 0
63+
);
64+
CREATE INDEX message_session_idx ON message (session_id);
65+
CREATE INDEX part_session_idx ON part (session_id);
66+
`)
67+
require.NoError(t, err, "create composite schema")
68+
return path, conn
69+
}
70+
71+
// TestFilterFreshWatermarkOnlySourcesRequiresLiveCapture pins the filter's
72+
// capture revalidation: the watermark comparison may only run against a
73+
// listing taken from a container that provably did not change between the
74+
// pre-listing capture and filtering. A commit landing inside that window can
75+
// advance a session past its listed watermark, and at classification time no
76+
// pass-level capture guard exists to catch it — if every source were
77+
// filtered, the pass would end before beginSQLiteContainerPass ran at all.
78+
func TestFilterFreshWatermarkOnlySourcesRequiresLiveCapture(t *testing.T) {
79+
dbPath, conn := newCompositeContainerTestDB(t)
80+
_, err := conn.Exec(
81+
"INSERT INTO session (id, project_id, time_created, time_updated)" +
82+
" VALUES ('ses-1', 'proj', 1779012000000, 1779012000000)",
83+
)
84+
require.NoError(t, err, "insert session row")
85+
86+
root := filepath.Dir(dbPath)
87+
provider, ok := parser.NewProvider(
88+
parser.AgentOpenCode,
89+
parser.ProviderConfig{Roots: []string{root}, Machine: "local"},
90+
)
91+
require.True(t, ok)
92+
sources, err := provider.SourcesForChangedPath(
93+
t.Context(), parser.ChangedPathRequest{
94+
Path: dbPath, WatchRoot: root, AllowWatermarkOnlySources: true,
95+
},
96+
)
97+
require.NoError(t, err)
98+
require.Len(t, sources, 1)
99+
_, watermarkOnly := parser.SourceWatermarkOnlyMTimeNS(sources[0])
100+
require.True(t, watermarkOnly,
101+
"the listing must be the bounded watermark-only form")
102+
103+
database := openTestDB(t)
104+
virtual := dbPath + "#ses-1"
105+
storedMtime := int64(1779012000000) * 1_000_000
106+
require.NoError(t, database.UpsertSession(db.Session{
107+
ID: "opencode:ses-1", Agent: "opencode", Project: "project",
108+
Machine: "local", FilePath: &virtual, FileMtime: &storedMtime,
109+
}))
110+
// UpsertSession seeds data_version 0 by design; stamp it as a completed
111+
// parse would, or the filter keeps the source for a version rewrite.
112+
require.NoError(t, database.SetSessionDataVersion(
113+
"opencode:ses-1", db.CurrentDataVersion(),
114+
))
115+
e := &Engine{db: database, machine: "local"}
116+
117+
pre, ok := statSQLiteContainerState(dbPath)
118+
require.True(t, ok, "capture must be readable")
119+
assert.Empty(t, e.filterFreshWatermarkOnlySources(
120+
t.Context(), dbPath, pre, true, sources,
121+
), "a covered source under a live capture is filtered")
122+
123+
// The write lands after the listing and the capture: the listed
124+
// watermark no longer reflects the container, so nothing may be
125+
// filtered against it.
126+
_, err = conn.Exec(
127+
"UPDATE session SET time_updated = 1779012999000 WHERE id = 'ses-1'",
128+
)
129+
require.NoError(t, err, "advance session row")
130+
assert.Len(t, e.filterFreshWatermarkOnlySources(
131+
t.Context(), dbPath, pre, true, sources,
132+
), 1, "a stale capture must keep every source for full fingerprinting")
133+
134+
assert.Len(t, e.filterFreshWatermarkOnlySources(
135+
t.Context(), dbPath, parser.SQLiteContainerState{}, false, sources,
136+
), 1, "a missing capture must keep every source")
137+
}
138+
28139
// TestSQLiteContainerPassPromotesOnlyPreDiscoveryCaptures pins the gate's
29140
// ordering invariant: the state promoted to trusted must have been captured
30141
// BEFORE discovery listed the container's sessions. Discovery reads the

0 commit comments

Comments
 (0)