Skip to content

Commit 34b3440

Browse files
wesmmaphew
wesm
and
maphew
committed
fix(artifact): re-dirty export queue across resync and origin adoption
CopySyncStateFrom now forces copied queue rows to pending so a resynced archive re-verifies every session: the origin gate keeps triggers silent in the rebuild's temp DB, so copied rows must carry the dirty flag themselves. Divergent origin adoption requeues all owned sessions with a generation bump instead of the INSERT OR IGNORE bootstrap, which would leave a fully acknowledged ledger empty under the new origin. A failed queue population now deletes the origin key when there was no previous origin, keeping the existence-gated triggers closed. The test-only queued-export iterator moved out of production code. Co-authored-by: maphew <maphew@gmail.com>
1 parent 4b9ff0c commit 34b3440

7 files changed

Lines changed: 325 additions & 82 deletions

File tree

‎internal/artifact/export.go‎

Lines changed: 0 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -52,66 +52,6 @@ type ExportResult struct {
5252
CheckpointSequence int
5353
}
5454

55-
type queuedArtifactExportStore interface {
56-
PendingArtifactExports(context.Context, int) ([]db.ArtifactExportQueueItem, error)
57-
GetSessionFull(context.Context, string) (*db.Session, error)
58-
GetAllMessages(context.Context, string) ([]db.Message, error)
59-
GetUsageEvents(context.Context, string) ([]db.UsageEvent, error)
60-
}
61-
62-
type queuedArtifactExport struct {
63-
Item db.ArtifactExportQueueItem
64-
Session *db.Session
65-
Messages []db.Message
66-
UsageEvents []db.UsageEvent
67-
}
68-
69-
// forEachQueuedArtifactExport loads only the bounded dirty batch and at most
70-
// one complete session body at a time. A missing session represents a pending
71-
// publication deletion and deliberately performs no message or usage reads.
72-
func forEachQueuedArtifactExport(
73-
ctx context.Context,
74-
store queuedArtifactExportStore,
75-
limit int,
76-
visit func(queuedArtifactExport) error,
77-
) error {
78-
if visit == nil {
79-
return errors.New("queued artifact export visitor is required")
80-
}
81-
items, err := store.PendingArtifactExports(ctx, limit)
82-
if err != nil {
83-
return fmt.Errorf("reading queued artifact exports: %w", err)
84-
}
85-
for _, item := range items {
86-
if err := ctx.Err(); err != nil {
87-
return err
88-
}
89-
work := queuedArtifactExport{Item: item}
90-
work.Session, err = store.GetSessionFull(ctx, item.SessionID)
91-
if err != nil {
92-
return fmt.Errorf("loading queued artifact session %s: %w", item.SessionID, err)
93-
}
94-
if work.Session != nil &&
95-
(work.Session.Machine != "local" || work.Session.DeletedAt != nil) {
96-
work.Session = nil
97-
}
98-
if work.Session != nil {
99-
work.Messages, err = store.GetAllMessages(ctx, item.SessionID)
100-
if err != nil {
101-
return fmt.Errorf("loading queued artifact messages %s: %w", item.SessionID, err)
102-
}
103-
work.UsageEvents, err = store.GetUsageEvents(ctx, item.SessionID)
104-
if err != nil {
105-
return fmt.Errorf("loading queued artifact usage %s: %w", item.SessionID, err)
106-
}
107-
}
108-
if err := visit(work); err != nil {
109-
return err
110-
}
111-
}
112-
return nil
113-
}
114-
11555
// ExportToStore publishes generation-guarded work into the canonical artifact
11656
// store. Immutable dependencies are created before their manifest, and each
11757
// bounded page's checkpoint is created last. Full mode may publish several

‎internal/artifact/export_helpers_test.go‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
package artifact
2+
3+
import (
4+
"context"
5+
"errors"
6+
"fmt"
7+
8+
"go.kenn.io/agentsview/internal/db"
9+
)
10+
11+
type queuedArtifactExportStore interface {
12+
PendingArtifactExports(context.Context, int) ([]db.ArtifactExportQueueItem, error)
13+
GetSessionFull(context.Context, string) (*db.Session, error)
14+
GetAllMessages(context.Context, string) ([]db.Message, error)
15+
GetUsageEvents(context.Context, string) ([]db.UsageEvent, error)
16+
}
17+
18+
type queuedArtifactExport struct {
19+
Item db.ArtifactExportQueueItem
20+
Session *db.Session
21+
Messages []db.Message
22+
UsageEvents []db.UsageEvent
23+
}
24+
25+
// forEachQueuedArtifactExport loads only the bounded dirty batch and at most
26+
// one complete session body at a time. A missing session represents a pending
27+
// publication deletion and deliberately performs no message or usage reads.
28+
func forEachQueuedArtifactExport(
29+
ctx context.Context,
30+
store queuedArtifactExportStore,
31+
limit int,
32+
visit func(queuedArtifactExport) error,
33+
) error {
34+
if visit == nil {
35+
return errors.New("queued artifact export visitor is required")
36+
}
37+
items, err := store.PendingArtifactExports(ctx, limit)
38+
if err != nil {
39+
return fmt.Errorf("reading queued artifact exports: %w", err)
40+
}
41+
for _, item := range items {
42+
if err := ctx.Err(); err != nil {
43+
return err
44+
}
45+
work := queuedArtifactExport{Item: item}
46+
work.Session, err = store.GetSessionFull(ctx, item.SessionID)
47+
if err != nil {
48+
return fmt.Errorf("loading queued artifact session %s: %w", item.SessionID, err)
49+
}
50+
if work.Session != nil &&
51+
(work.Session.Machine != "local" || work.Session.DeletedAt != nil) {
52+
work.Session = nil
53+
}
54+
if work.Session != nil {
55+
work.Messages, err = store.GetAllMessages(ctx, item.SessionID)
56+
if err != nil {
57+
return fmt.Errorf("loading queued artifact messages %s: %w", item.SessionID, err)
58+
}
59+
work.UsageEvents, err = store.GetUsageEvents(ctx, item.SessionID)
60+
if err != nil {
61+
return fmt.Errorf("loading queued artifact usage %s: %w", item.SessionID, err)
62+
}
63+
}
64+
if err := visit(work); err != nil {
65+
return err
66+
}
67+
}
68+
return nil
69+
}

‎internal/artifact/origin.go‎

Lines changed: 33 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -11,19 +11,34 @@ import (
1111
"go.kenn.io/agentsview/internal/db"
1212
)
1313

14-
// bootstrapExportQueue is a test seam for injecting bootstrap failures.
15-
var bootstrapExportQueue = (*db.DB).BootstrapArtifactExportQueue
14+
// bootstrapExportQueue and requeueExportQueue are test seams for injecting
15+
// queue-population failures. Bootstrap enqueues each session once (INSERT OR
16+
// IGNORE); requeue force-dirties every owned session for a divergent origin.
17+
var (
18+
bootstrapExportQueue = (*db.DB).BootstrapArtifactExportQueue
19+
requeueExportQueue = (*db.DB).RequeueAllArtifactExports
20+
)
1621

17-
// bootstrapQueueForOrigin populates the export queue after origin persists.
18-
// On failure it rolls the stored origin back to previous so a retry re-runs
19-
// the bootstrap instead of fast-pathing to success with an unpopulated queue.
20-
func bootstrapQueueForOrigin(database *db.DB, origin, previous string) error {
21-
err := bootstrapExportQueue(database)
22+
// populateQueueForOrigin runs populate after the origin persists. On failure it
23+
// rolls the stored origin back to previous so a retry re-runs population instead
24+
// of fast-pathing to success with an unpopulated queue. When previous is empty
25+
// the origin key is deleted rather than set to an empty value, because the
26+
// export gates test key existence, not the stored value.
27+
func populateQueueForOrigin(
28+
database *db.DB, populate func(*db.DB) error, origin, previous string,
29+
) error {
30+
err := populate(database)
2231
if err == nil {
2332
return nil
2433
}
25-
err = fmt.Errorf("bootstrapping export queue for origin %s: %w", origin, err)
26-
if rollbackErr := database.SetSyncState(originStateKey, previous); rollbackErr != nil {
34+
err = fmt.Errorf("populating export queue for origin %s: %w", origin, err)
35+
var rollbackErr error
36+
if previous == "" {
37+
rollbackErr = database.DeleteSyncState(originStateKey)
38+
} else {
39+
rollbackErr = database.SetSyncState(originStateKey, previous)
40+
}
41+
if rollbackErr != nil {
2742
return errors.Join(err,
2843
fmt.Errorf("rolling back artifact origin: %w", rollbackErr))
2944
}
@@ -49,7 +64,7 @@ func EnsureOrigin(database *db.DB) (string, error) {
4964
if err := database.SetSyncState(originStateKey, origin); err != nil {
5065
return "", fmt.Errorf("persisting artifact origin: %w", err)
5166
}
52-
if err := bootstrapQueueForOrigin(database, origin, ""); err != nil {
67+
if err := populateQueueForOrigin(database, bootstrapExportQueue, origin, ""); err != nil {
5368
return "", err
5469
}
5570
return origin, nil
@@ -74,7 +89,14 @@ func AdoptOrigin(database *db.DB, origin string) error {
7489
if err := database.SetSyncState(originStateKey, origin); err != nil {
7590
return fmt.Errorf("persisting artifact origin: %w", err)
7691
}
77-
return bootstrapQueueForOrigin(database, origin, existing)
92+
// A divergent adoption replaces an established origin whose sessions may
93+
// already be acknowledged, so bootstrap's INSERT OR IGNORE would leave the
94+
// new origin's ledger empty. Force-requeue every owned session instead.
95+
// First-time adoption keeps the cheaper bootstrap.
96+
if existing != "" {
97+
return populateQueueForOrigin(database, requeueExportQueue, origin, existing)
98+
}
99+
return populateQueueForOrigin(database, bootstrapExportQueue, origin, existing)
78100
}
79101

80102
// StoredOrigin returns the persisted origin ID without creating one.

‎internal/artifact/origin_test.go‎

Lines changed: 58 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -128,30 +128,81 @@ func TestEnsureOriginRollsBackWhenBootstrapFails(t *testing.T) {
128128
require.NoError(t, err)
129129
assert.Empty(t, stored, "failed bootstrap must roll the origin back")
130130

131+
// The rollback deletes the origin key entirely rather than writing an empty
132+
// value, so the export gate (which tests key existence) stays closed. A
133+
// session written after the failed creation must not enqueue.
134+
seedSession(t, database, "sess-2", "alpha")
135+
gated, err := database.PendingArtifactExports(t.Context(), 10)
136+
require.NoError(t, err)
137+
require.Empty(t, gated, "failed origin creation must leave the export gate closed")
138+
131139
bootstrapExportQueue = (*db.DB).BootstrapArtifactExportQueue
132140
origin, err := EnsureOrigin(database)
133141
require.NoError(t, err)
134142
require.NotEmpty(t, origin, "retry after rollback must re-run creation")
135143
pending, err := database.PendingArtifactExports(t.Context(), 10)
136144
require.NoError(t, err)
137-
require.Len(t, pending, 1, "retry must re-run the bootstrap")
138-
assert.Equal(t, "sess-1", pending[0].SessionID)
145+
require.Len(t, pending, 2, "retry must re-run the bootstrap for every pre-existing session")
146+
assert.ElementsMatch(t, []string{"sess-1", "sess-2"}, []string{
147+
pending[0].SessionID, pending[1].SessionID,
148+
})
139149
}
140150

141-
func TestAdoptOriginRestoresPreviousOriginWhenBootstrapFails(t *testing.T) {
151+
func TestAdoptOriginRestoresPreviousOriginWhenRequeueFails(t *testing.T) {
142152
database := testDB(t)
143153
require.NoError(t, AdoptOrigin(database, "before-a1b2c3"))
144154

145-
injected := errors.New("bootstrap exploded")
146-
bootstrapExportQueue = func(*db.DB) error { return injected }
147-
t.Cleanup(func() { bootstrapExportQueue = (*db.DB).BootstrapArtifactExportQueue })
155+
injected := errors.New("requeue exploded")
156+
requeueExportQueue = func(*db.DB) error { return injected }
157+
t.Cleanup(func() { requeueExportQueue = (*db.DB).RequeueAllArtifactExports })
148158

159+
// Adopting a divergent origin over an established one routes through the
160+
// requeue path, not bootstrap.
149161
err := AdoptOrigin(database, "after-d4e5f6")
150162
require.ErrorIs(t, err, injected)
151163
stored, err := StoredOrigin(database)
152164
require.NoError(t, err)
153165
assert.Equal(t, "before-a1b2c3", stored,
154-
"failed adoption must restore the previous origin")
166+
"failed divergent adoption must restore the previous origin")
167+
}
168+
169+
// TestAdoptOriginRequeuesAllExportsOnDivergentAdoption covers the divergent
170+
// adoption path: when a new origin replaces an established one whose sessions
171+
// are already acknowledged, INSERT OR IGNORE bootstrap would leave the ledger
172+
// empty, so every owned session must be force-requeued with a bumped
173+
// generation.
174+
func TestAdoptOriginRequeuesAllExportsOnDivergentAdoption(t *testing.T) {
175+
database := testDB(t)
176+
require.NoError(t, AdoptOrigin(database, "origin-a1b2c3"))
177+
seedSession(t, database, "sess-1", "alpha")
178+
seedSession(t, database, "sess-2", "alpha")
179+
180+
ctx := t.Context()
181+
pending, err := database.PendingArtifactExports(ctx, 10)
182+
require.NoError(t, err)
183+
require.Len(t, pending, 2)
184+
genBefore := map[string]int64{}
185+
for _, item := range pending {
186+
genBefore[item.SessionID] = item.Generation
187+
}
188+
189+
// Simulate the prior origin having fully published every session.
190+
require.NoError(t, database.AcknowledgeArtifactExports(ctx, pending))
191+
drained, err := database.PendingArtifactExports(ctx, 10)
192+
require.NoError(t, err)
193+
require.Empty(t, drained)
194+
195+
require.NoError(t, AdoptOrigin(database, "origin-d4e5f6"))
196+
pending, err = database.PendingArtifactExports(ctx, 10)
197+
require.NoError(t, err)
198+
require.Len(t, pending, 2, "divergent adoption re-verifies every owned session")
199+
assert.ElementsMatch(t, []string{"sess-1", "sess-2"}, []string{
200+
pending[0].SessionID, pending[1].SessionID,
201+
})
202+
for _, item := range pending {
203+
assert.Greater(t, item.Generation, genBefore[item.SessionID],
204+
"divergent adoption must bump the generation of every requeued session")
205+
}
155206
}
156207

157208
// TestAdoptOriginBootstrapsPreExistingLocalSessions mirrors the EnsureOrigin

0 commit comments

Comments
 (0)