fix(daemon): keep snapshot generations contiguous
- commit
- 6de1185be55ff1d49e0fe948cc191b1c298d9b7c
- author
- brice · · 2026-07-22 03:54 UTC
- parent
- 5c87cd6
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +135 -21
M internal/usecase/daemon/session.go +9 -8
| @@ -53,20 +53,21 @@ type session struct { | |||
| 53 | 53 | mruAt atomic.Uint64 | |
| 54 | 54 | snapDirty atomic.Bool | |
| 55 | 55 | snapEligible atomic.Bool | |
| 56 | - | // snapshotMu serializes the dirty generation with worker completion. It is | |
| 57 | - | // intentionally independent from mu: persistence never holds session state | |
| 58 | - | // locks while encoding or writing. | |
| 59 | - | snapshotMu sync.Mutex | |
| 60 | - | snapshotGeneration uint64 | |
| 61 | - | snapshotPublishedGeneration uint64 | |
| 62 | - | snapshotNextEligibleAt time.Time | |
| 56 | + | // snapshotMu serializes mutation revisions and repository publication | |
| 57 | + | // generations with worker completion. It is intentionally independent from | |
| 58 | + | // mu: persistence never holds session state locks while encoding or writing. | |
| 59 | + | snapshotMu sync.Mutex | |
| 60 | + | snapshotGeneration uint64 // newest mutation revision | |
| 61 | + | snapshotPublishedGeneration uint64 // newest repository generation | |
| 62 | + | snapshotPublishedMutationRevision uint64 | |
| 63 | + | snapshotNextEligibleAt time.Time | |
| 63 | 64 | // The coordinator state below is guarded by snapshotMu. A capture can be | |
| 64 | 65 | // queued globally or in flight, never more than one of each per session. | |
| 65 | 66 | // Quarantine cancels the session publication context before destructive | |
| 66 | 67 | // repository operations and publicationDone joins a started publication. | |
| 67 | 68 | snapshotPending bool | |
| 68 | 69 | snapshotPendingCaptures uint | |
| 69 | - | // snapshotForcedGeneration is the newest dirty generation a forced | |
| 70 | + | // snapshotForcedGeneration is the newest mutation revision a forced | |
| 70 | 71 | // checkpoint must publish. It survives an older routine capture so worker | |
| 71 | 72 | // completion can enqueue exactly one forced successor. | |
| 72 | 73 | snapshotForcedGeneration uint64 | |
| @@ -53,20 +53,21 @@ type session struct { | |||||
| 53 | mruAt atomic.Uint64 | 53 | mruAt atomic.Uint64 | ||
| 54 | snapDirty atomic.Bool | 54 | snapDirty atomic.Bool | ||
| 55 | snapEligible atomic.Bool | 55 | snapEligible atomic.Bool | ||
| 56 | - | // snapshotMu serializes the dirty generation with worker completion. It is | 56 | + | // snapshotMu serializes mutation revisions and repository publication |
| 57 | - | // intentionally independent from mu: persistence never holds session state | 57 | + | // generations with worker completion. It is intentionally independent from |
| 58 | - | // locks while encoding or writing. | 58 | + | // mu: persistence never holds session state locks while encoding or writing. |
| 59 | - | snapshotMu sync.Mutex | 59 | + | snapshotMu sync.Mutex |
| 60 | - | snapshotGeneration uint64 | 60 | + | snapshotGeneration uint64 // newest mutation revision |
| 61 | - | snapshotPublishedGeneration uint64 | 61 | + | snapshotPublishedGeneration uint64 // newest repository generation |
| 62 | - | snapshotNextEligibleAt time.Time | 62 | + | snapshotPublishedMutationRevision uint64 |
| 63 | + | snapshotNextEligibleAt time.Time | |||
| 63 | // The coordinator state below is guarded by snapshotMu. A capture can be | 64 | // The coordinator state below is guarded by snapshotMu. A capture can be | ||
| 64 | // queued globally or in flight, never more than one of each per session. | 65 | // queued globally or in flight, never more than one of each per session. | ||
| 65 | // Quarantine cancels the session publication context before destructive | 66 | // Quarantine cancels the session publication context before destructive | ||
| 66 | // repository operations and publicationDone joins a started publication. | 67 | // repository operations and publicationDone joins a started publication. | ||
| 67 | snapshotPending bool | 68 | snapshotPending bool | ||
| 68 | snapshotPendingCaptures uint | 69 | snapshotPendingCaptures uint | ||
| 69 | - | // snapshotForcedGeneration is the newest dirty generation a forced | 70 | + | // snapshotForcedGeneration is the newest mutation revision a forced |
| 70 | // checkpoint must publish. It survives an older routine capture so worker | 71 | // checkpoint must publish. It survives an older routine capture so worker | ||
| 71 | // completion can enqueue exactly one forced successor. | 72 | // completion can enqueue exactly one forced successor. | ||
| 72 | snapshotForcedGeneration uint64 | 73 | snapshotForcedGeneration uint64 | ||
M internal/usecase/daemon/snapshot.go +2 -1
| @@ -34,7 +34,8 @@ const ( | |||
| 34 | 34 | type snapshotCapture struct { | |
| 35 | 35 | session *session | |
| 36 | 36 | attemptKind snapshotAttemptKind | |
| 37 | - | generation uint64 | |
| 37 | + | generation uint64 // repository publication generation | |
| 38 | + | mutationRevision uint64 | |
| 38 | 39 | name string | |
| 39 | 40 | createdAt uint64 | |
| 40 | 41 | active uint16 | |
| @@ -34,7 +34,8 @@ const ( | |||||
| 34 | type snapshotCapture struct { | 34 | type snapshotCapture struct { | ||
| 35 | session *session | 35 | session *session | ||
| 36 | attemptKind snapshotAttemptKind | 36 | attemptKind snapshotAttemptKind | ||
| 37 | - | generation uint64 | 37 | + | generation uint64 // repository publication generation |
| 38 | + | mutationRevision uint64 | |||
| 38 | name string | 39 | name string | ||
| 39 | createdAt uint64 | 40 | createdAt uint64 | ||
| 40 | active uint16 | 41 | active uint16 | ||
M internal/usecase/daemon/snapshot_coordinator.go +16 -7
| @@ -90,6 +90,7 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC | |||
| 90 | 90 | // dirty and wake the scheduler. | |
| 91 | 91 | sess.snapshotGeneration = 1 | |
| 92 | 92 | sess.snapshotPublishedGeneration = 0 | |
| 93 | + | sess.snapshotPublishedMutationRevision = 0 | |
| 93 | 94 | sess.snapshotForcedGeneration = 0 | |
| 94 | 95 | sess.snapshotNextEligibleAt = time.Time{} | |
| 95 | 96 | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | |
| @@ -142,9 +143,9 @@ func (sess *session) signalSnapshotChangedLocked() { | |||
| 142 | 143 | } | |
| 143 | 144 | ||
| 144 | 145 | // waitForSnapshotGeneration waits only for the bounded terminal checkpoint | |
| 145 | - | // deadline. It observes successful publication, rather than dirty state, so a | |
| 146 | - | // concurrent later mutation cannot turn a completed forced checkpoint into a | |
| 147 | - | // false timeout. | |
| 146 | + | // deadline. It observes successful publication of the requested mutation | |
| 147 | + | // revision, rather than dirty state, so a concurrent later mutation cannot | |
| 148 | + | // turn a completed forced checkpoint into a false timeout. | |
| 148 | 149 | func (d *Daemon) waitForSnapshotGeneration(sess *session, generation uint64) bool { | |
| 149 | 150 | if d == nil || sess == nil || generation == 0 { | |
| 150 | 151 | return true | |
| @@ -166,7 +167,7 @@ func (d *Daemon) waitForSnapshotGenerationWithDeadline(sess *session, generation | |||
| 166 | 167 | } | |
| 167 | 168 | for { | |
| 168 | 169 | sess.snapshotMu.Lock() | |
| 169 | - | published := sess.snapshotPublishedGeneration >= generation | |
| 170 | + | published := sess.snapshotPublishedMutationRevision >= generation | |
| 170 | 171 | changed := sess.snapshotChangeLocked() | |
| 171 | 172 | sess.snapshotMu.Unlock() | |
| 172 | 173 | if published { | |
| @@ -229,23 +230,31 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) | |||
| 229 | 230 | // attempt. A forced teardown checkpoint deliberately leaves it alone. | |
| 230 | 231 | capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval) | |
| 231 | 232 | } | |
| 233 | + | mutationRevision := capture.mutationRevision | |
| 234 | + | // Older unit fixtures created captures before mutation revisions were | |
| 235 | + | // recorded explicitly. Production captures always set it at scheduling; | |
| 236 | + | // retaining this fallback keeps those fixtures representative. | |
| 237 | + | if mutationRevision == 0 { | |
| 238 | + | mutationRevision = capture.generation | |
| 239 | + | } | |
| 232 | 240 | if succeeded { | |
| 233 | 241 | capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation) | |
| 242 | + | capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision) | |
| 234 | 243 | } | |
| 235 | - | if succeeded && capture.session.snapshotGeneration == capture.generation { | |
| 244 | + | if succeeded && capture.session.snapshotGeneration == mutationRevision { | |
| 236 | 245 | capture.session.snapDirty.Store(false) | |
| 237 | 246 | } else if !succeeded { | |
| 238 | 247 | capture.session.snapDirty.Store(true) | |
| 239 | 248 | } | |
| 240 | 249 | if succeeded && capture.session.snapshotForcedGeneration != 0 && | |
| 241 | - | capture.generation >= capture.session.snapshotForcedGeneration { | |
| 250 | + | mutationRevision >= capture.session.snapshotForcedGeneration { | |
| 242 | 251 | capture.session.snapshotForcedGeneration = 0 | |
| 243 | 252 | } | |
| 244 | 253 | // A failed forced publication remains dirty for ordinary retry rather | |
| 245 | 254 | // than recursively creating forced jobs. A later forced request records | |
| 246 | 255 | // a newer intent and can create its one bounded successor. | |
| 247 | 256 | shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 && | |
| 248 | - | capture.generation < capture.session.snapshotForcedGeneration && | |
| 257 | + | mutationRevision < capture.session.snapshotForcedGeneration && | |
| 249 | 258 | capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil && | |
| 250 | 259 | (succeeded || capture.attemptKind != snapshotAttemptForced) | |
| 251 | 260 | capture.session.signalSnapshotChangedLocked() | |
| @@ -90,6 +90,7 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC | |||||
| 90 | // dirty and wake the scheduler. | 90 | // dirty and wake the scheduler. | ||
| 91 | sess.snapshotGeneration = 1 | 91 | sess.snapshotGeneration = 1 | ||
| 92 | sess.snapshotPublishedGeneration = 0 | 92 | sess.snapshotPublishedGeneration = 0 | ||
| 93 | + | sess.snapshotPublishedMutationRevision = 0 | |||
| 93 | sess.snapshotForcedGeneration = 0 | 94 | sess.snapshotForcedGeneration = 0 | ||
| 94 | sess.snapshotNextEligibleAt = time.Time{} | 95 | sess.snapshotNextEligibleAt = time.Time{} | ||
| 95 | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | 96 | sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit) | ||
| @@ -142,9 +143,9 @@ func (sess *session) signalSnapshotChangedLocked() { | |||||
| 142 | } | 143 | } | ||
| 143 | 144 | ||||
| 144 | // waitForSnapshotGeneration waits only for the bounded terminal checkpoint | 145 | // waitForSnapshotGeneration waits only for the bounded terminal checkpoint | ||
| 145 | - | // deadline. It observes successful publication, rather than dirty state, so a | 146 | + | // deadline. It observes successful publication of the requested mutation |
| 146 | - | // concurrent later mutation cannot turn a completed forced checkpoint into a | 147 | + | // revision, rather than dirty state, so a concurrent later mutation cannot |
| 147 | - | // false timeout. | 148 | + | // turn a completed forced checkpoint into a false timeout. |
| 148 | func (d *Daemon) waitForSnapshotGeneration(sess *session, generation uint64) bool { | 149 | func (d *Daemon) waitForSnapshotGeneration(sess *session, generation uint64) bool { | ||
| 149 | if d == nil || sess == nil || generation == 0 { | 150 | if d == nil || sess == nil || generation == 0 { | ||
| 150 | return true | 151 | return true | ||
| @@ -166,7 +167,7 @@ func (d *Daemon) waitForSnapshotGenerationWithDeadline(sess *session, generation | |||||
| 166 | } | 167 | } | ||
| 167 | for { | 168 | for { | ||
| 168 | sess.snapshotMu.Lock() | 169 | sess.snapshotMu.Lock() | ||
| 169 | - | published := sess.snapshotPublishedGeneration >= generation | 170 | + | published := sess.snapshotPublishedMutationRevision >= generation |
| 170 | changed := sess.snapshotChangeLocked() | 171 | changed := sess.snapshotChangeLocked() | ||
| 171 | sess.snapshotMu.Unlock() | 172 | sess.snapshotMu.Unlock() | ||
| 172 | if published { | 173 | if published { | ||
| @@ -229,23 +230,31 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) | |||||
| 229 | // attempt. A forced teardown checkpoint deliberately leaves it alone. | 230 | // attempt. A forced teardown checkpoint deliberately leaves it alone. | ||
| 230 | capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval) | 231 | capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval) | ||
| 231 | } | 232 | } | ||
| 233 | + | mutationRevision := capture.mutationRevision | |||
| 234 | + | // Older unit fixtures created captures before mutation revisions were | |||
| 235 | + | // recorded explicitly. Production captures always set it at scheduling; | |||
| 236 | + | // retaining this fallback keeps those fixtures representative. | |||
| 237 | + | if mutationRevision == 0 { | |||
| 238 | + | mutationRevision = capture.generation | |||
| 239 | + | } | |||
| 232 | if succeeded { | 240 | if succeeded { | ||
| 233 | capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation) | 241 | capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation) | ||
| 242 | + | capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision) | |||
| 234 | } | 243 | } | ||
| 235 | - | if succeeded && capture.session.snapshotGeneration == capture.generation { | 244 | + | if succeeded && capture.session.snapshotGeneration == mutationRevision { |
| 236 | capture.session.snapDirty.Store(false) | 245 | capture.session.snapDirty.Store(false) | ||
| 237 | } else if !succeeded { | 246 | } else if !succeeded { | ||
| 238 | capture.session.snapDirty.Store(true) | 247 | capture.session.snapDirty.Store(true) | ||
| 239 | } | 248 | } | ||
| 240 | if succeeded && capture.session.snapshotForcedGeneration != 0 && | 249 | if succeeded && capture.session.snapshotForcedGeneration != 0 && | ||
| 241 | - | capture.generation >= capture.session.snapshotForcedGeneration { | 250 | + | mutationRevision >= capture.session.snapshotForcedGeneration { |
| 242 | capture.session.snapshotForcedGeneration = 0 | 251 | capture.session.snapshotForcedGeneration = 0 | ||
| 243 | } | 252 | } | ||
| 244 | // A failed forced publication remains dirty for ordinary retry rather | 253 | // A failed forced publication remains dirty for ordinary retry rather | ||
| 245 | // than recursively creating forced jobs. A later forced request records | 254 | // than recursively creating forced jobs. A later forced request records | ||
| 246 | // a newer intent and can create its one bounded successor. | 255 | // a newer intent and can create its one bounded successor. | ||
| 247 | shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 && | 256 | shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 && | ||
| 248 | - | capture.generation < capture.session.snapshotForcedGeneration && | 257 | + | mutationRevision < capture.session.snapshotForcedGeneration && |
| 249 | capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil && | 258 | capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil && | ||
| 250 | (succeeded || capture.attemptKind != snapshotAttemptForced) | 259 | (succeeded || capture.attemptKind != snapshotAttemptForced) | ||
| 251 | capture.session.signalSnapshotChangedLocked() | 260 | capture.session.signalSnapshotChangedLocked() | ||
A internal/usecase/daemon/snapshot_generation_test.go +80 -0
| @@ -0,0 +1,80 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "errors" | |
| 6 | + | "sync/atomic" | |
| 7 | + | "testing" | |
| 8 | + | "time" | |
| 9 | + | ||
| 10 | + | "github.com/stretchr/testify/mock" | |
| 11 | + | "github.com/stretchr/testify/require" | |
| 12 | + | ||
| 13 | + | "github.com/bnema/vev/internal/ports" | |
| 14 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 15 | + | ) | |
| 16 | + | ||
| 17 | + | func TestSnapshotMultipleMutationsPublishNextRepositoryGeneration(t *testing.T) { | |
| 18 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 19 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |
| 20 | + | WithSnapshotRepository(repository, nil)(d) | |
| 21 | + | startSnapshotEncodeWorker(t, d) | |
| 22 | + | ||
| 23 | + | publications := make(chan ports.SnapshotPublication, 1) | |
| 24 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |
| 25 | + | publications <- publication | |
| 26 | + | return nil | |
| 27 | + | }).Once() | |
| 28 | + | ||
| 29 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |
| 30 | + | markSnapshotDirty(sess) | |
| 31 | + | markSnapshotDirty(sess) | |
| 32 | + | markSnapshotDirty(sess) | |
| 33 | + | require.True(t, d.scheduleSnapshot(sess)) | |
| 34 | + | ||
| 35 | + | select { | |
| 36 | + | case publication := <-publications: | |
| 37 | + | require.Equal(t, uint64(1), publication.Generation) | |
| 38 | + | case <-time.After(testWaitTimeout): | |
| 39 | + | t.Fatal("snapshot was not published") | |
| 40 | + | } | |
| 41 | + | awaitSnapshotClean(t, sess) | |
| 42 | + | } | |
| 43 | + | ||
| 44 | + | func TestSnapshotRetryKeepsRepositoryGeneration(t *testing.T) { | |
| 45 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 46 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |
| 47 | + | WithSnapshotRepository(repository, nil)(d) | |
| 48 | + | startSnapshotEncodeWorker(t, d) | |
| 49 | + | ||
| 50 | + | publications := make(chan ports.SnapshotPublication, 2) | |
| 51 | + | var attempts atomic.Int32 | |
| 52 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |
| 53 | + | publications <- publication | |
| 54 | + | if attempts.Add(1) == 1 { | |
| 55 | + | return errors.New("temporary repository failure") | |
| 56 | + | } | |
| 57 | + | return nil | |
| 58 | + | }).Twice() | |
| 59 | + | ||
| 60 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |
| 61 | + | markSnapshotDirty(sess) | |
| 62 | + | require.True(t, d.scheduleSnapshot(sess)) | |
| 63 | + | select { | |
| 64 | + | case publication := <-publications: | |
| 65 | + | require.Equal(t, uint64(1), publication.Generation) | |
| 66 | + | case <-time.After(testWaitTimeout): | |
| 67 | + | t.Fatal("first snapshot was not published") | |
| 68 | + | } | |
| 69 | + | awaitSnapshotIdle(t, sess) | |
| 70 | + | require.True(t, sess.snapDirty.Load()) | |
| 71 | + | ||
| 72 | + | require.True(t, d.scheduleFinalSnapshot(sess)) | |
| 73 | + | select { | |
| 74 | + | case publication := <-publications: | |
| 75 | + | require.Equal(t, uint64(1), publication.Generation) | |
| 76 | + | case <-time.After(testWaitTimeout): | |
| 77 | + | t.Fatal("retry snapshot was not published") | |
| 78 | + | } | |
| 79 | + | awaitSnapshotClean(t, sess) | |
| 80 | + | } | |
| @@ -0,0 +1,80 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "errors" | |||
| 6 | + | "sync/atomic" | |||
| 7 | + | "testing" | |||
| 8 | + | "time" | |||
| 9 | + | ||||
| 10 | + | "github.com/stretchr/testify/mock" | |||
| 11 | + | "github.com/stretchr/testify/require" | |||
| 12 | + | ||||
| 13 | + | "github.com/bnema/vev/internal/ports" | |||
| 14 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |||
| 15 | + | ) | |||
| 16 | + | ||||
| 17 | + | func TestSnapshotMultipleMutationsPublishNextRepositoryGeneration(t *testing.T) { | |||
| 18 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 19 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |||
| 20 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 21 | + | startSnapshotEncodeWorker(t, d) | |||
| 22 | + | ||||
| 23 | + | publications := make(chan ports.SnapshotPublication, 1) | |||
| 24 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |||
| 25 | + | publications <- publication | |||
| 26 | + | return nil | |||
| 27 | + | }).Once() | |||
| 28 | + | ||||
| 29 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |||
| 30 | + | markSnapshotDirty(sess) | |||
| 31 | + | markSnapshotDirty(sess) | |||
| 32 | + | markSnapshotDirty(sess) | |||
| 33 | + | require.True(t, d.scheduleSnapshot(sess)) | |||
| 34 | + | ||||
| 35 | + | select { | |||
| 36 | + | case publication := <-publications: | |||
| 37 | + | require.Equal(t, uint64(1), publication.Generation) | |||
| 38 | + | case <-time.After(testWaitTimeout): | |||
| 39 | + | t.Fatal("snapshot was not published") | |||
| 40 | + | } | |||
| 41 | + | awaitSnapshotClean(t, sess) | |||
| 42 | + | } | |||
| 43 | + | ||||
| 44 | + | func TestSnapshotRetryKeepsRepositoryGeneration(t *testing.T) { | |||
| 45 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |||
| 46 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |||
| 47 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 48 | + | startSnapshotEncodeWorker(t, d) | |||
| 49 | + | ||||
| 50 | + | publications := make(chan ports.SnapshotPublication, 2) | |||
| 51 | + | var attempts atomic.Int32 | |||
| 52 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error { | |||
| 53 | + | publications <- publication | |||
| 54 | + | if attempts.Add(1) == 1 { | |||
| 55 | + | return errors.New("temporary repository failure") | |||
| 56 | + | } | |||
| 57 | + | return nil | |||
| 58 | + | }).Twice() | |||
| 59 | + | ||||
| 60 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |||
| 61 | + | markSnapshotDirty(sess) | |||
| 62 | + | require.True(t, d.scheduleSnapshot(sess)) | |||
| 63 | + | select { | |||
| 64 | + | case publication := <-publications: | |||
| 65 | + | require.Equal(t, uint64(1), publication.Generation) | |||
| 66 | + | case <-time.After(testWaitTimeout): | |||
| 67 | + | t.Fatal("first snapshot was not published") | |||
| 68 | + | } | |||
| 69 | + | awaitSnapshotIdle(t, sess) | |||
| 70 | + | require.True(t, sess.snapDirty.Load()) | |||
| 71 | + | ||||
| 72 | + | require.True(t, d.scheduleFinalSnapshot(sess)) | |||
| 73 | + | select { | |||
| 74 | + | case publication := <-publications: | |||
| 75 | + | require.Equal(t, uint64(1), publication.Generation) | |||
| 76 | + | case <-time.After(testWaitTimeout): | |||
| 77 | + | t.Fatal("retry snapshot was not published") | |||
| 78 | + | } | |||
| 79 | + | awaitSnapshotClean(t, sess) | |||
| 80 | + | } | |||
M internal/usecase/daemon/snapshot_queue.go +13 -1
| @@ -41,7 +41,18 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||
| 41 | 41 | if sess.snapshotPublicationContext == nil { | |
| 42 | 42 | sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background()) | |
| 43 | 43 | } | |
| 44 | - | generation := sess.snapshotGeneration | |
| 44 | + | // A capture's repository generation advances only from the last successful | |
| 45 | + | // repository publication. Mutations are independently versioned, so several | |
| 46 | + | // changes coalesced before this checkpoint still publish the immediate next | |
| 47 | + | // repository generation and a failed attempt retries that same generation. | |
| 48 | + | generation := sess.snapshotPublishedGeneration + 1 | |
| 49 | + | mutationRevision := sess.snapshotGeneration | |
| 50 | + | // A restored or test-created dirty session can predate mutation revision | |
| 51 | + | // tracking. Normalize that state into the first revision before capture. | |
| 52 | + | if mutationRevision == 0 { | |
| 53 | + | mutationRevision = 1 | |
| 54 | + | sess.snapshotGeneration = mutationRevision | |
| 55 | + | } | |
| 45 | 56 | publicationContext := sess.snapshotPublicationContext | |
| 46 | 57 | sess.snapshotPendingCaptures++ | |
| 47 | 58 | sess.snapshotPending = true | |
| @@ -50,6 +61,7 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||
| 50 | 61 | ||
| 51 | 62 | capture, ok := d.captureSnapshotState(sess, generation) | |
| 52 | 63 | if capture != nil { | |
| 64 | + | capture.mutationRevision = mutationRevision | |
| 53 | 65 | capture.attemptKind = kind | |
| 54 | 66 | capture.publicationContext = publicationContext | |
| 55 | 67 | } | |
| @@ -41,7 +41,18 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||||
| 41 | if sess.snapshotPublicationContext == nil { | 41 | if sess.snapshotPublicationContext == nil { | ||
| 42 | sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background()) | 42 | sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background()) | ||
| 43 | } | 43 | } | ||
| 44 | - | generation := sess.snapshotGeneration | 44 | + | // A capture's repository generation advances only from the last successful |
| 45 | + | // repository publication. Mutations are independently versioned, so several | |||
| 46 | + | // changes coalesced before this checkpoint still publish the immediate next | |||
| 47 | + | // repository generation and a failed attempt retries that same generation. | |||
| 48 | + | generation := sess.snapshotPublishedGeneration + 1 | |||
| 49 | + | mutationRevision := sess.snapshotGeneration | |||
| 50 | + | // A restored or test-created dirty session can predate mutation revision | |||
| 51 | + | // tracking. Normalize that state into the first revision before capture. | |||
| 52 | + | if mutationRevision == 0 { | |||
| 53 | + | mutationRevision = 1 | |||
| 54 | + | sess.snapshotGeneration = mutationRevision | |||
| 55 | + | } | |||
| 45 | publicationContext := sess.snapshotPublicationContext | 56 | publicationContext := sess.snapshotPublicationContext | ||
| 46 | sess.snapshotPendingCaptures++ | 57 | sess.snapshotPendingCaptures++ | ||
| 47 | sess.snapshotPending = true | 58 | sess.snapshotPending = true | ||
| @@ -50,6 +61,7 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||||
| 50 | 61 | ||||
| 51 | capture, ok := d.captureSnapshotState(sess, generation) | 62 | capture, ok := d.captureSnapshotState(sess, generation) | ||
| 52 | if capture != nil { | 63 | if capture != nil { | ||
| 64 | + | capture.mutationRevision = mutationRevision | |||
| 53 | capture.attemptKind = kind | 65 | capture.attemptKind = kind | ||
| 54 | capture.publicationContext = publicationContext | 66 | capture.publicationContext = publicationContext | ||
| 55 | } | 67 | } | ||
M internal/usecase/daemon/snapshot_restore.go +4 -1
| @@ -64,7 +64,7 @@ func validateRestoreSessionSnapshot(snap snapcodec.Session) error { | |||
| 64 | 64 | ||
| 65 | 65 | // restoreSession owns the restored tabs until registration succeeds. Every | |
| 66 | 66 | // unsuccessful path closes their PTYs and cancels their shared session context. | |
| 67 | - | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) error { | |
| 67 | + | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64) error { | |
| 68 | 68 | if err := ctx.Err(); err != nil { | |
| 69 | 69 | return err | |
| 70 | 70 | } | |
| @@ -98,6 +98,9 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err | |||
| 98 | 98 | }() | |
| 99 | 99 | ||
| 100 | 100 | sess := d.newRestoredSession(snap, sctx, cancel, opened) | |
| 101 | + | // The loaded manifest is the repository head for this name. Future dirty | |
| 102 | + | // checkpoints must continue from it rather than reuse generation one. | |
| 103 | + | sess.snapshotPublishedGeneration = repositoryGeneration | |
| 101 | 104 | registered, err := d.persistAndRegisterRestoredSession(ctx, sess) | |
| 102 | 105 | if err != nil || !registered { | |
| 103 | 106 | return err | |
| @@ -64,7 +64,7 @@ func validateRestoreSessionSnapshot(snap snapcodec.Session) error { | |||||
| 64 | 64 | ||||
| 65 | // restoreSession owns the restored tabs until registration succeeds. Every | 65 | // restoreSession owns the restored tabs until registration succeeds. Every | ||
| 66 | // unsuccessful path closes their PTYs and cancels their shared session context. | 66 | // unsuccessful path closes their PTYs and cancels their shared session context. | ||
| 67 | - | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) error { | 67 | + | func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session, repositoryGeneration uint64) error { |
| 68 | if err := ctx.Err(); err != nil { | 68 | if err := ctx.Err(); err != nil { | ||
| 69 | return err | 69 | return err | ||
| 70 | } | 70 | } | ||
| @@ -98,6 +98,9 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err | |||||
| 98 | }() | 98 | }() | ||
| 99 | 99 | ||||
| 100 | sess := d.newRestoredSession(snap, sctx, cancel, opened) | 100 | sess := d.newRestoredSession(snap, sctx, cancel, opened) | ||
| 101 | + | // The loaded manifest is the repository head for this name. Future dirty | |||
| 102 | + | // checkpoints must continue from it rather than reuse generation one. | |||
| 103 | + | sess.snapshotPublishedGeneration = repositoryGeneration | |||
| 101 | registered, err := d.persistAndRegisterRestoredSession(ctx, sess) | 104 | registered, err := d.persistAndRegisterRestoredSession(ctx, sess) | ||
| 102 | if err != nil || !registered { | 105 | if err != nil || !registered { | ||
| 103 | return err | 106 | return err | ||
M internal/usecase/daemon/snapshot_restore_incremental.go +1 -1
| @@ -39,7 +39,7 @@ func (d *Daemon) restoreIncrementalSnapshots(ctx context.Context) { | |||
| 39 | 39 | d.snapshotRestoreFailure(name, err) | |
| 40 | 40 | continue | |
| 41 | 41 | } | |
| 42 | - | if err := d.restoreSession(ctx, snapshot); err != nil { | |
| 42 | + | if err := d.restoreSession(ctx, snapshot, generation.Generation); err != nil { | |
| 43 | 43 | if ctx.Err() == nil { | |
| 44 | 44 | d.snapshotRestoreFailure(name, err) | |
| 45 | 45 | } | |
| @@ -39,7 +39,7 @@ func (d *Daemon) restoreIncrementalSnapshots(ctx context.Context) { | |||||
| 39 | d.snapshotRestoreFailure(name, err) | 39 | d.snapshotRestoreFailure(name, err) | ||
| 40 | continue | 40 | continue | ||
| 41 | } | 41 | } | ||
| 42 | - | if err := d.restoreSession(ctx, snapshot); err != nil { | 42 | + | if err := d.restoreSession(ctx, snapshot, generation.Generation); err != nil { |
| 43 | if ctx.Err() == nil { | 43 | if ctx.Err() == nil { | ||
| 44 | d.snapshotRestoreFailure(name, err) | 44 | d.snapshotRestoreFailure(name, err) | ||
| 45 | } | 45 | } | ||
M internal/usecase/daemon/snapshot_restore_test.go +8 -0
| @@ -197,6 +197,7 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) { | |||
| 197 | 197 | d.mu.Unlock() | |
| 198 | 198 | require.NotNil(t, restored) | |
| 199 | 199 | require.False(t, restored.snapDirty.Load(), "a loaded generation must begin clean") | |
| 200 | + | require.Equal(t, uint64(9), restored.snapshotPublishedGeneration, "the loaded manifest is the repository generation head") | |
| 200 | 201 | require.Equal(t, 0, restored.active) | |
| 201 | 202 | require.Equal(t, "/snapshot/cwd", restored.cwd) | |
| 202 | 203 | require.Len(t, restored.tabs, 1) | |
| @@ -211,6 +212,13 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) { | |||
| 211 | 212 | require.Equal(t, defaultScrollbackRows, pane.history.Cap()) | |
| 212 | 213 | require.Equal(t, defaultScrollbackCells, pane.history.CellCap()) | |
| 213 | 214 | pane.mu.Unlock() | |
| 215 | + | ||
| 216 | + | startSnapshotEncodeWorker(t, d) | |
| 217 | + | markSnapshotDirty(restored) | |
| 218 | + | require.True(t, d.scheduleSnapshot(restored)) | |
| 219 | + | awaitSnapshotClean(t, restored) | |
| 220 | + | require.Len(t, repository.publishes, 1) | |
| 221 | + | require.Equal(t, uint64(10), repository.publishes[0].Generation, "a restored session must continue the concrete repository generation stream") | |
| 214 | 222 | } | |
| 215 | 223 | ||
| 216 | 224 | func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) { | |
| @@ -197,6 +197,7 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) { | |||||
| 197 | d.mu.Unlock() | 197 | d.mu.Unlock() | ||
| 198 | require.NotNil(t, restored) | 198 | require.NotNil(t, restored) | ||
| 199 | require.False(t, restored.snapDirty.Load(), "a loaded generation must begin clean") | 199 | require.False(t, restored.snapDirty.Load(), "a loaded generation must begin clean") | ||
| 200 | + | require.Equal(t, uint64(9), restored.snapshotPublishedGeneration, "the loaded manifest is the repository generation head") | |||
| 200 | require.Equal(t, 0, restored.active) | 201 | require.Equal(t, 0, restored.active) | ||
| 201 | require.Equal(t, "/snapshot/cwd", restored.cwd) | 202 | require.Equal(t, "/snapshot/cwd", restored.cwd) | ||
| 202 | require.Len(t, restored.tabs, 1) | 203 | require.Len(t, restored.tabs, 1) | ||
| @@ -211,6 +212,13 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) { | |||||
| 211 | require.Equal(t, defaultScrollbackRows, pane.history.Cap()) | 212 | require.Equal(t, defaultScrollbackRows, pane.history.Cap()) | ||
| 212 | require.Equal(t, defaultScrollbackCells, pane.history.CellCap()) | 213 | require.Equal(t, defaultScrollbackCells, pane.history.CellCap()) | ||
| 213 | pane.mu.Unlock() | 214 | pane.mu.Unlock() | ||
| 215 | + | ||||
| 216 | + | startSnapshotEncodeWorker(t, d) | |||
| 217 | + | markSnapshotDirty(restored) | |||
| 218 | + | require.True(t, d.scheduleSnapshot(restored)) | |||
| 219 | + | awaitSnapshotClean(t, restored) | |||
| 220 | + | require.Len(t, repository.publishes, 1) | |||
| 221 | + | require.Equal(t, uint64(10), repository.publishes[0].Generation, "a restored session must continue the concrete repository generation stream") | |||
| 214 | } | 222 | } | ||
| 215 | 223 | ||||
| 216 | func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) { | 224 | func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) { | ||
M internal/usecase/daemon/snapshot_scheduler_test.go +2 -2
| @@ -352,7 +352,7 @@ func TestMultipleForcedSnapshotsCoalesceToOneSuccessor(t *testing.T) { | |||
| 352 | 352 | ||
| 353 | 353 | awaitSnapshotClean(t, sess) | |
| 354 | 354 | require.Equal(t, uint64(1), (<-publications).Generation) | |
| 355 | - | require.Equal(t, uint64(3), (<-publications).Generation) | |
| 355 | + | require.Equal(t, uint64(2), (<-publications).Generation) | |
| 356 | 356 | sess.snapshotMu.Lock() | |
| 357 | 357 | require.False(t, sess.snapshotPending) | |
| 358 | 358 | require.Nil(t, sess.snapshotQueuedCapture) | |
| @@ -911,7 +911,7 @@ func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) { | |||
| 911 | 911 | close(release) | |
| 912 | 912 | ||
| 913 | 913 | gotGenerations := []uint64{(<-publications).Generation, (<-publications).Generation, (<-publications).Generation} | |
| 914 | - | require.ElementsMatch(t, []uint64{1, 1, 3}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded") | |
| 914 | + | require.ElementsMatch(t, []uint64{1, 1, 2}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded") | |
| 915 | 915 | awaitSnapshotCleanSignal(inFlight) | |
| 916 | 916 | } | |
| 917 | 917 | ||
| @@ -352,7 +352,7 @@ func TestMultipleForcedSnapshotsCoalesceToOneSuccessor(t *testing.T) { | |||||
| 352 | 352 | ||||
| 353 | awaitSnapshotClean(t, sess) | 353 | awaitSnapshotClean(t, sess) | ||
| 354 | require.Equal(t, uint64(1), (<-publications).Generation) | 354 | require.Equal(t, uint64(1), (<-publications).Generation) | ||
| 355 | - | require.Equal(t, uint64(3), (<-publications).Generation) | 355 | + | require.Equal(t, uint64(2), (<-publications).Generation) |
| 356 | sess.snapshotMu.Lock() | 356 | sess.snapshotMu.Lock() | ||
| 357 | require.False(t, sess.snapshotPending) | 357 | require.False(t, sess.snapshotPending) | ||
| 358 | require.Nil(t, sess.snapshotQueuedCapture) | 358 | require.Nil(t, sess.snapshotQueuedCapture) | ||
| @@ -911,7 +911,7 @@ func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) { | |||||
| 911 | close(release) | 911 | close(release) | ||
| 912 | 912 | ||||
| 913 | gotGenerations := []uint64{(<-publications).Generation, (<-publications).Generation, (<-publications).Generation} | 913 | gotGenerations := []uint64{(<-publications).Generation, (<-publications).Generation, (<-publications).Generation} | ||
| 914 | - | require.ElementsMatch(t, []uint64{1, 1, 3}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded") | 914 | + | require.ElementsMatch(t, []uint64{1, 1, 2}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded") |
| 915 | awaitSnapshotCleanSignal(inFlight) | 915 | awaitSnapshotCleanSignal(inFlight) | ||
| 916 | } | 916 | } | ||
| 917 | 917 | ||||