fix(daemon): schedule repository checkpoints
- commit
- 8aaf1e5733f30644a7b6af3bb23bd86d2e47d508
- author
- brice · · 2026-07-21 22:01 UTC
- parent
- cd4708e
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
8 files · +387 -55
M internal/usecase/daemon/daemon.go +34 -24
| @@ -129,8 +129,12 @@ type Daemon struct { | |||
| 129 | 129 | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | |
| 130 | 130 | // independent of pane state and is bounded to prevent checkpoint history | |
| 131 | 131 | // from becoming unbounded daemon memory. | |
| 132 | - | snapshotChunkCache *snapshotChunkCache | |
| 133 | - | snapshotJobs chan *snapshotCapture | |
| 132 | + | snapshotChunkCache *snapshotChunkCache | |
| 133 | + | snapshotJobs chan *snapshotCapture | |
| 134 | + | // snapshotWake wakes the repository scheduler when a session becomes dirty | |
| 135 | + | // or an attempt completes. It is never closed and producers only send | |
| 136 | + | // non-blockingly. | |
| 137 | + | snapshotWake chan struct{} | |
| 134 | 138 | snapshotWorkerMu sync.Mutex | |
| 135 | 139 | snapshotWorkerID uint64 | |
| 136 | 140 | snapshotWorkerCtx context.Context | |
| @@ -141,28 +145,33 @@ type Daemon struct { | |||
| 141 | 145 | // snapshotFinalJobs coalesces terminal captures by session when the bounded | |
| 142 | 146 | // regular queue is full. It retains at most snapshotFinalQueueCapacity named | |
| 143 | 147 | // sessions, each with only its newest terminal state while the worker blocks. | |
| 144 | - | snapshotFinalJobs map[*session]*snapshotCapture | |
| 145 | - | snapshotFinalOrder []*session | |
| 146 | - | snapshotWorkerClosing bool | |
| 147 | - | snapshotWorkerInFlight *snapshotCapture | |
| 148 | - | restoreDone chan struct{} | |
| 149 | - | restoreOnce sync.Once | |
| 150 | - | procCwd func(int) (string, error) | |
| 151 | - | procComm func(int) (string, error) | |
| 152 | - | procArgv func(int) ([]string, error) | |
| 153 | - | procGroupArgv func(int, int) ([]string, error) | |
| 154 | - | dirOrHome func(string) string | |
| 155 | - | bindings atomic.Pointer[keys.Bindings] | |
| 156 | - | codeOverrides atomic.Pointer[map[string]string] | |
| 157 | - | restoreProcessAllowlist atomic.Pointer[map[string]struct{}] | |
| 158 | - | floatingConfig atomic.Pointer[domain.FloatingConfig] | |
| 159 | - | copyConfig atomic.Pointer[domain.CopyConfig] | |
| 160 | - | paletteConfig atomic.Pointer[domain.PaletteConfig] | |
| 161 | - | tabsConfig atomic.Pointer[domain.TabsConfig] | |
| 162 | - | themeConfig atomic.Pointer[themeConfigSnapshot] | |
| 163 | - | barScripts *barScriptState | |
| 164 | - | notices *noticeCenter | |
| 165 | - | resumeParkGrace time.Duration | |
| 148 | + | snapshotFinalJobs map[*session]*snapshotCapture | |
| 149 | + | snapshotFinalOrder []*session | |
| 150 | + | snapshotWorkerClosing bool | |
| 151 | + | snapshotWorkerInFlight *snapshotCapture | |
| 152 | + | // snapshotNoticeMu guards the active global persistence failure signature. | |
| 153 | + | // It is separate from snapshotWorkerMu so notice routing cannot block a | |
| 154 | + | // producer or a repository worker. | |
| 155 | + | snapshotNoticeMu sync.Mutex | |
| 156 | + | snapshotActiveFailureSignature string | |
| 157 | + | restoreDone chan struct{} | |
| 158 | + | restoreOnce sync.Once | |
| 159 | + | procCwd func(int) (string, error) | |
| 160 | + | procComm func(int) (string, error) | |
| 161 | + | procArgv func(int) ([]string, error) | |
| 162 | + | procGroupArgv func(int, int) ([]string, error) | |
| 163 | + | dirOrHome func(string) string | |
| 164 | + | bindings atomic.Pointer[keys.Bindings] | |
| 165 | + | codeOverrides atomic.Pointer[map[string]string] | |
| 166 | + | restoreProcessAllowlist atomic.Pointer[map[string]struct{}] | |
| 167 | + | floatingConfig atomic.Pointer[domain.FloatingConfig] | |
| 168 | + | copyConfig atomic.Pointer[domain.CopyConfig] | |
| 169 | + | paletteConfig atomic.Pointer[domain.PaletteConfig] | |
| 170 | + | tabsConfig atomic.Pointer[domain.TabsConfig] | |
| 171 | + | themeConfig atomic.Pointer[themeConfigSnapshot] | |
| 172 | + | barScripts *barScriptState | |
| 173 | + | notices *noticeCenter | |
| 174 | + | resumeParkGrace time.Duration | |
| 166 | 175 | // tempDir overrides os.TempDir() for clipboard-image-transfer writes | |
| 167 | 176 | // (see clipboard.go); empty means use os.TempDir(). | |
| 168 | 177 | tempDir string | |
| @@ -383,6 +392,7 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt | |||
| 383 | 392 | snapshotMarshal: snapcodec.Marshal, | |
| 384 | 393 | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), | |
| 385 | 394 | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | |
| 395 | + | snapshotWake: make(chan struct{}, 1), | |
| 386 | 396 | notices: newNoticeCenter(), | |
| 387 | 397 | resumeParkGrace: defaultResumeParkGrace, | |
| 388 | 398 | barScripts: &barScriptState{ | |
| @@ -129,8 +129,12 @@ type Daemon struct { | |||||
| 129 | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | 129 | // snapshotChunkCache contains only encoded immutable sealed chunks. It is | ||
| 130 | // independent of pane state and is bounded to prevent checkpoint history | 130 | // independent of pane state and is bounded to prevent checkpoint history | ||
| 131 | // from becoming unbounded daemon memory. | 131 | // from becoming unbounded daemon memory. | ||
| 132 | - | snapshotChunkCache *snapshotChunkCache | 132 | + | snapshotChunkCache *snapshotChunkCache |
| 133 | - | snapshotJobs chan *snapshotCapture | 133 | + | snapshotJobs chan *snapshotCapture |
| 134 | + | // snapshotWake wakes the repository scheduler when a session becomes dirty | |||
| 135 | + | // or an attempt completes. It is never closed and producers only send | |||
| 136 | + | // non-blockingly. | |||
| 137 | + | snapshotWake chan struct{} | |||
| 134 | snapshotWorkerMu sync.Mutex | 138 | snapshotWorkerMu sync.Mutex | ||
| 135 | snapshotWorkerID uint64 | 139 | snapshotWorkerID uint64 | ||
| 136 | snapshotWorkerCtx context.Context | 140 | snapshotWorkerCtx context.Context | ||
| @@ -141,28 +145,33 @@ type Daemon struct { | |||||
| 141 | // snapshotFinalJobs coalesces terminal captures by session when the bounded | 145 | // snapshotFinalJobs coalesces terminal captures by session when the bounded | ||
| 142 | // regular queue is full. It retains at most snapshotFinalQueueCapacity named | 146 | // regular queue is full. It retains at most snapshotFinalQueueCapacity named | ||
| 143 | // sessions, each with only its newest terminal state while the worker blocks. | 147 | // sessions, each with only its newest terminal state while the worker blocks. | ||
| 144 | - | snapshotFinalJobs map[*session]*snapshotCapture | 148 | + | snapshotFinalJobs map[*session]*snapshotCapture |
| 145 | - | snapshotFinalOrder []*session | 149 | + | snapshotFinalOrder []*session |
| 146 | - | snapshotWorkerClosing bool | 150 | + | snapshotWorkerClosing bool |
| 147 | - | snapshotWorkerInFlight *snapshotCapture | 151 | + | snapshotWorkerInFlight *snapshotCapture |
| 148 | - | restoreDone chan struct{} | 152 | + | // snapshotNoticeMu guards the active global persistence failure signature. |
| 149 | - | restoreOnce sync.Once | 153 | + | // It is separate from snapshotWorkerMu so notice routing cannot block a |
| 150 | - | procCwd func(int) (string, error) | 154 | + | // producer or a repository worker. |
| 151 | - | procComm func(int) (string, error) | 155 | + | snapshotNoticeMu sync.Mutex |
| 152 | - | procArgv func(int) ([]string, error) | 156 | + | snapshotActiveFailureSignature string |
| 153 | - | procGroupArgv func(int, int) ([]string, error) | 157 | + | restoreDone chan struct{} |
| 154 | - | dirOrHome func(string) string | 158 | + | restoreOnce sync.Once |
| 155 | - | bindings atomic.Pointer[keys.Bindings] | 159 | + | procCwd func(int) (string, error) |
| 156 | - | codeOverrides atomic.Pointer[map[string]string] | 160 | + | procComm func(int) (string, error) |
| 157 | - | restoreProcessAllowlist atomic.Pointer[map[string]struct{}] | 161 | + | procArgv func(int) ([]string, error) |
| 158 | - | floatingConfig atomic.Pointer[domain.FloatingConfig] | 162 | + | procGroupArgv func(int, int) ([]string, error) |
| 159 | - | copyConfig atomic.Pointer[domain.CopyConfig] | 163 | + | dirOrHome func(string) string |
| 160 | - | paletteConfig atomic.Pointer[domain.PaletteConfig] | 164 | + | bindings atomic.Pointer[keys.Bindings] |
| 161 | - | tabsConfig atomic.Pointer[domain.TabsConfig] | 165 | + | codeOverrides atomic.Pointer[map[string]string] |
| 162 | - | themeConfig atomic.Pointer[themeConfigSnapshot] | 166 | + | restoreProcessAllowlist atomic.Pointer[map[string]struct{}] |
| 163 | - | barScripts *barScriptState | 167 | + | floatingConfig atomic.Pointer[domain.FloatingConfig] |
| 164 | - | notices *noticeCenter | 168 | + | copyConfig atomic.Pointer[domain.CopyConfig] |
| 165 | - | resumeParkGrace time.Duration | 169 | + | paletteConfig atomic.Pointer[domain.PaletteConfig] |
| 170 | + | tabsConfig atomic.Pointer[domain.TabsConfig] | |||
| 171 | + | themeConfig atomic.Pointer[themeConfigSnapshot] | |||
| 172 | + | barScripts *barScriptState | |||
| 173 | + | notices *noticeCenter | |||
| 174 | + | resumeParkGrace time.Duration | |||
| 166 | // tempDir overrides os.TempDir() for clipboard-image-transfer writes | 175 | // tempDir overrides os.TempDir() for clipboard-image-transfer writes | ||
| 167 | // (see clipboard.go); empty means use os.TempDir(). | 176 | // (see clipboard.go); empty means use os.TempDir(). | ||
| 168 | tempDir string | 177 | tempDir string | ||
| @@ -383,6 +392,7 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt | |||||
| 383 | snapshotMarshal: snapcodec.Marshal, | 392 | snapshotMarshal: snapcodec.Marshal, | ||
| 384 | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), | 393 | snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit), | ||
| 385 | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | 394 | snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity), | ||
| 395 | + | snapshotWake: make(chan struct{}, 1), | |||
| 386 | notices: newNoticeCenter(), | 396 | notices: newNoticeCenter(), | ||
| 387 | resumeParkGrace: defaultResumeParkGrace, | 397 | resumeParkGrace: defaultResumeParkGrace, | ||
| 388 | barScripts: &barScriptState{ | 398 | barScripts: &barScriptState{ | ||
M internal/usecase/daemon/notify.go +25 -0
| @@ -61,6 +61,31 @@ func (nc *noticeCenter) record(n domain.Notification) domain.Notification { | |||
| 61 | 61 | return n | |
| 62 | 62 | } | |
| 63 | 63 | ||
| 64 | + | // recordSnapshotFailure coalesces a repeated stable snapshot failure in the | |
| 65 | + | // history. Details contains the signature rather than raw error text, so this | |
| 66 | + | // comparison cannot be affected by volatile paths or snapshot content. | |
| 67 | + | func (nc *noticeCenter) recordSnapshotFailure(n domain.Notification) (domain.Notification, bool) { | |
| 68 | + | if n.Count == 0 { | |
| 69 | + | n.Count = 1 | |
| 70 | + | } | |
| 71 | + | nc.mu.Lock() | |
| 72 | + | defer nc.mu.Unlock() | |
| 73 | + | for i := len(nc.ring) - 1; i >= 0; i-- { | |
| 74 | + | previous := &nc.ring[i] | |
| 75 | + | if previous.Code != domain.NoticeSnapshotWrite || previous.SessionID != n.SessionID || previous.Details != n.Details { | |
| 76 | + | continue | |
| 77 | + | } | |
| 78 | + | previous.Count += n.Count | |
| 79 | + | previous.Time = n.Time | |
| 80 | + | return *previous, true | |
| 81 | + | } | |
| 82 | + | nc.ring = append(nc.ring, n) | |
| 83 | + | if len(nc.ring) > noticeHistoryCap { | |
| 84 | + | nc.ring = nc.ring[len(nc.ring)-noticeHistoryCap:] | |
| 85 | + | } | |
| 86 | + | return n, false | |
| 87 | + | } | |
| 88 | + | ||
| 64 | 89 | func (nc *noticeCenter) history() []domain.Notification { | |
| 65 | 90 | nc.mu.Lock() | |
| 66 | 91 | defer nc.mu.Unlock() | |
| @@ -61,6 +61,31 @@ func (nc *noticeCenter) record(n domain.Notification) domain.Notification { | |||||
| 61 | return n | 61 | return n | ||
| 62 | } | 62 | } | ||
| 63 | 63 | ||||
| 64 | + | // recordSnapshotFailure coalesces a repeated stable snapshot failure in the | |||
| 65 | + | // history. Details contains the signature rather than raw error text, so this | |||
| 66 | + | // comparison cannot be affected by volatile paths or snapshot content. | |||
| 67 | + | func (nc *noticeCenter) recordSnapshotFailure(n domain.Notification) (domain.Notification, bool) { | |||
| 68 | + | if n.Count == 0 { | |||
| 69 | + | n.Count = 1 | |||
| 70 | + | } | |||
| 71 | + | nc.mu.Lock() | |||
| 72 | + | defer nc.mu.Unlock() | |||
| 73 | + | for i := len(nc.ring) - 1; i >= 0; i-- { | |||
| 74 | + | previous := &nc.ring[i] | |||
| 75 | + | if previous.Code != domain.NoticeSnapshotWrite || previous.SessionID != n.SessionID || previous.Details != n.Details { | |||
| 76 | + | continue | |||
| 77 | + | } | |||
| 78 | + | previous.Count += n.Count | |||
| 79 | + | previous.Time = n.Time | |||
| 80 | + | return *previous, true | |||
| 81 | + | } | |||
| 82 | + | nc.ring = append(nc.ring, n) | |||
| 83 | + | if len(nc.ring) > noticeHistoryCap { | |||
| 84 | + | nc.ring = nc.ring[len(nc.ring)-noticeHistoryCap:] | |||
| 85 | + | } | |||
| 86 | + | return n, false | |||
| 87 | + | } | |||
| 88 | + | ||||
| 64 | func (nc *noticeCenter) history() []domain.Notification { | 89 | func (nc *noticeCenter) history() []domain.Notification { | ||
| 65 | nc.mu.Lock() | 90 | nc.mu.Lock() | ||
| 66 | defer nc.mu.Unlock() | 91 | defer nc.mu.Unlock() | ||
M internal/usecase/daemon/session.go +24 -15
| @@ -55,11 +55,19 @@ type session struct { | |||
| 55 | 55 | // snapshotMu serializes the dirty generation with worker completion. It is | |
| 56 | 56 | // intentionally independent from mu: persistence never holds session state | |
| 57 | 57 | // locks while encoding or writing. | |
| 58 | - | snapshotMu sync.Mutex | |
| 59 | - | snapshotGeneration uint64 | |
| 60 | - | snapshotPending bool | |
| 61 | - | snapshotPendingCaptures uint | |
| 62 | - | snapshotChanged chan struct{} | |
| 58 | + | snapshotMu sync.Mutex | |
| 59 | + | snapshotGeneration uint64 | |
| 60 | + | snapshotCapturedGeneration uint64 | |
| 61 | + | snapshotNextEligibleAt time.Time | |
| 62 | + | snapshotAttempted bool | |
| 63 | + | snapshotAttemptKind snapshotAttemptKind | |
| 64 | + | snapshotPending bool | |
| 65 | + | snapshotPendingCaptures uint | |
| 66 | + | // snapshotFailureSig is the stable class of the active failed publication. | |
| 67 | + | // It is guarded by snapshotMu and deliberately never contains error text. | |
| 68 | + | snapshotFailureSig string | |
| 69 | + | snapshotChanged chan struct{} | |
| 70 | + | snapshotWake chan struct{} | |
| 63 | 71 | // syncGen makes synchronized-output watchdog generations unique across all | |
| 64 | 72 | // panes in this session. | |
| 65 | 73 | syncGen atomic.Uint64 | |
| @@ -193,16 +201,17 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||
| 193 | 201 | lastUsedSeq = stopped.lastUsedSeq | |
| 194 | 202 | } | |
| 195 | 203 | sess := &session{ | |
| 196 | - | id: id, | |
| 197 | - | name: name, | |
| 198 | - | ephemeral: ephemeral, | |
| 199 | - | ctx: sctx, | |
| 200 | - | cancel: cancel, | |
| 201 | - | tabs: tabs, | |
| 202 | - | cwd: cwd, | |
| 203 | - | terminal: term, | |
| 204 | - | env: env, | |
| 205 | - | createdAt: createdAt, | |
| 204 | + | id: id, | |
| 205 | + | name: name, | |
| 206 | + | ephemeral: ephemeral, | |
| 207 | + | ctx: sctx, | |
| 208 | + | cancel: cancel, | |
| 209 | + | tabs: tabs, | |
| 210 | + | cwd: cwd, | |
| 211 | + | terminal: term, | |
| 212 | + | env: env, | |
| 213 | + | createdAt: createdAt, | |
| 214 | + | snapshotWake: d.snapshotWake, | |
| 206 | 215 | } | |
| 207 | 216 | if lastUsedSeq > 0 { | |
| 208 | 217 | sess.mruAt.Store(lastUsedSeq) | |
| @@ -55,11 +55,19 @@ type session struct { | |||||
| 55 | // snapshotMu serializes the dirty generation with worker completion. It is | 55 | // snapshotMu serializes the dirty generation with worker completion. It is | ||
| 56 | // intentionally independent from mu: persistence never holds session state | 56 | // intentionally independent from mu: persistence never holds session state | ||
| 57 | // locks while encoding or writing. | 57 | // locks while encoding or writing. | ||
| 58 | - | snapshotMu sync.Mutex | 58 | + | snapshotMu sync.Mutex |
| 59 | - | snapshotGeneration uint64 | 59 | + | snapshotGeneration uint64 |
| 60 | - | snapshotPending bool | 60 | + | snapshotCapturedGeneration uint64 |
| 61 | - | snapshotPendingCaptures uint | 61 | + | snapshotNextEligibleAt time.Time |
| 62 | - | snapshotChanged chan struct{} | 62 | + | snapshotAttempted bool |
| 63 | + | snapshotAttemptKind snapshotAttemptKind | |||
| 64 | + | snapshotPending bool | |||
| 65 | + | snapshotPendingCaptures uint | |||
| 66 | + | // snapshotFailureSig is the stable class of the active failed publication. | |||
| 67 | + | // It is guarded by snapshotMu and deliberately never contains error text. | |||
| 68 | + | snapshotFailureSig string | |||
| 69 | + | snapshotChanged chan struct{} | |||
| 70 | + | snapshotWake chan struct{} | |||
| 63 | // syncGen makes synchronized-output watchdog generations unique across all | 71 | // syncGen makes synchronized-output watchdog generations unique across all | ||
| 64 | // panes in this session. | 72 | // panes in this session. | ||
| 65 | syncGen atomic.Uint64 | 73 | syncGen atomic.Uint64 | ||
| @@ -193,16 +201,17 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz | |||||
| 193 | lastUsedSeq = stopped.lastUsedSeq | 201 | lastUsedSeq = stopped.lastUsedSeq | ||
| 194 | } | 202 | } | ||
| 195 | sess := &session{ | 203 | sess := &session{ | ||
| 196 | - | id: id, | 204 | + | id: id, |
| 197 | - | name: name, | 205 | + | name: name, |
| 198 | - | ephemeral: ephemeral, | 206 | + | ephemeral: ephemeral, |
| 199 | - | ctx: sctx, | 207 | + | ctx: sctx, |
| 200 | - | cancel: cancel, | 208 | + | cancel: cancel, |
| 201 | - | tabs: tabs, | 209 | + | tabs: tabs, |
| 202 | - | cwd: cwd, | 210 | + | cwd: cwd, |
| 203 | - | terminal: term, | 211 | + | terminal: term, |
| 204 | - | env: env, | 212 | + | env: env, |
| 205 | - | createdAt: createdAt, | 213 | + | createdAt: createdAt, |
| 214 | + | snapshotWake: d.snapshotWake, | |||
| 206 | } | 215 | } | ||
| 207 | if lastUsedSeq > 0 { | 216 | if lastUsedSeq > 0 { | ||
| 208 | sess.mruAt.Store(lastUsedSeq) | 217 | sess.mruAt.Store(lastUsedSeq) | ||
M internal/usecase/daemon/snapshot.go +191 -14
| @@ -2,6 +2,7 @@ package daemon | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 5 | 6 | "fmt" | |
| 6 | 7 | "math" | |
| 7 | 8 | "sync" | |
| @@ -22,14 +23,25 @@ const snapshotQueueCapacity = 1 | |||
| 22 | 23 | // (and left dirty) rather than silently dropping a persistence acknowledgement. | |
| 23 | 24 | const snapshotFinalQueueCapacity = 32 | |
| 24 | 25 | ||
| 26 | + | // snapshotAttemptKind distinguishes rate-limited routine publications from | |
| 27 | + | // mandatory terminal checkpoints. Forced attempts never consume the routine | |
| 28 | + | // completion window. | |
| 29 | + | type snapshotAttemptKind uint8 | |
| 30 | + | ||
| 31 | + | const ( | |
| 32 | + | snapshotAttemptRoutine snapshotAttemptKind = iota | |
| 33 | + | snapshotAttemptForced | |
| 34 | + | ) | |
| 35 | + | ||
| 25 | 36 | type snapshotCapture struct { | |
| 26 | - | session *session | |
| 27 | - | generation uint64 | |
| 28 | - | name string | |
| 29 | - | createdAt uint64 | |
| 30 | - | active uint16 | |
| 31 | - | tabs []snapshotCaptureTab | |
| 32 | - | finishOnce sync.Once | |
| 37 | + | session *session | |
| 38 | + | attemptKind snapshotAttemptKind | |
| 39 | + | generation uint64 | |
| 40 | + | name string | |
| 41 | + | createdAt uint64 | |
| 42 | + | active uint16 | |
| 43 | + | tabs []snapshotCaptureTab | |
| 44 | + | finishOnce sync.Once | |
| 33 | 45 | } | |
| 34 | 46 | ||
| 35 | 47 | type snapshotCaptureTab struct { | |
| @@ -57,6 +69,24 @@ type snapshotCapturePane struct { | |||
| 57 | 69 | process *snapcodec.Process | |
| 58 | 70 | } | |
| 59 | 71 | ||
| 72 | + | // snapshotFailureSignature classifies a persistence failure without retaining | |
| 73 | + | // error text. Error strings often contain object content or filesystem paths, | |
| 74 | + | // neither of which is stable enough for deduplication or suitable for a global | |
| 75 | + | // notice history. | |
| 76 | + | func snapshotFailureSignature(phase string, err error) string { | |
| 77 | + | if err == nil { | |
| 78 | + | return phase + ":none" | |
| 79 | + | } | |
| 80 | + | switch { | |
| 81 | + | case errors.Is(err, context.Canceled): | |
| 82 | + | return phase + ":canceled" | |
| 83 | + | case errors.Is(err, context.DeadlineExceeded): | |
| 84 | + | return phase + ":deadline" | |
| 85 | + | default: | |
| 86 | + | return fmt.Sprintf("%s:%T", phase, err) | |
| 87 | + | } | |
| 88 | + | } | |
| 89 | + | ||
| 60 | 90 | const ( | |
| 61 | 91 | // Routine checkpoints are rate limited from completion; forced teardown | |
| 62 | 92 | // checkpoints use the same bounded worker but never consume this window. | |
| @@ -266,7 +296,7 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err | |||
| 266 | 296 | return nil | |
| 267 | 297 | } | |
| 268 | 298 | createdAt := int64(snap.CreatedAt) | |
| 269 | - | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt} | |
| 299 | + | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt, snapshotWake: d.snapshotWake} | |
| 270 | 300 | sess.snapEligible.Store(true) | |
| 271 | 301 | if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 { | |
| 272 | 302 | sess.cwd = snap.Tabs[0].Panes[0].Cwd | |
| @@ -308,7 +338,14 @@ func markSnapshotDirty(sess *session) { | |||
| 308 | 338 | sess.snapshotGeneration++ | |
| 309 | 339 | sess.snapDirty.Store(true) | |
| 310 | 340 | sess.signalSnapshotChangedLocked() | |
| 341 | + | wake := sess.snapshotWake | |
| 311 | 342 | sess.snapshotMu.Unlock() | |
| 343 | + | if wake != nil { | |
| 344 | + | select { | |
| 345 | + | case wake <- struct{}{}: | |
| 346 | + | default: | |
| 347 | + | } | |
| 348 | + | } | |
| 312 | 349 | } | |
| 313 | 350 | ||
| 314 | 351 | // snapshotChangeLocked returns a channel closed by the next snapshot state | |
| @@ -383,18 +420,29 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||
| 383 | 420 | if d == nil || sess == nil || !d.snapsEnabled || (d.snapshotRepository == nil && d.snaps == nil) || !sess.snapEligible.Load() { | |
| 384 | 421 | return true | |
| 385 | 422 | } | |
| 423 | + | kind := snapshotAttemptRoutine | |
| 424 | + | if final { | |
| 425 | + | kind = snapshotAttemptForced | |
| 426 | + | } | |
| 386 | 427 | sess.snapshotMu.Lock() | |
| 387 | - | if !sess.snapDirty.Load() || (!final && sess.snapshotPending) { | |
| 428 | + | routineRateLimited := d.snapshotRepository != nil | |
| 429 | + | if !sess.snapDirty.Load() || (!final && (sess.snapshotPending || (routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)))) { | |
| 388 | 430 | sess.snapshotMu.Unlock() | |
| 389 | 431 | return true | |
| 390 | 432 | } | |
| 391 | 433 | generation := sess.snapshotGeneration | |
| 434 | + | sess.snapshotCapturedGeneration = generation | |
| 435 | + | sess.snapshotAttempted = true | |
| 436 | + | sess.snapshotAttemptKind = kind | |
| 392 | 437 | sess.snapshotPendingCaptures++ | |
| 393 | 438 | sess.snapshotPending = true | |
| 394 | 439 | sess.signalSnapshotChangedLocked() | |
| 395 | 440 | sess.snapshotMu.Unlock() | |
| 396 | 441 | ||
| 397 | 442 | capture, ok := d.captureSnapshotState(sess, generation) | |
| 443 | + | if capture != nil { | |
| 444 | + | capture.attemptKind = kind | |
| 445 | + | } | |
| 398 | 446 | if !ok { | |
| 399 | 447 | d.finishSnapshotCapture(capture, false) | |
| 400 | 448 | return false | |
| @@ -556,8 +604,11 @@ func (d *Daemon) encodeSnapshotCapture(capture *snapshotCapture) ([]byte, error) | |||
| 556 | 604 | paneIDs := make(map[layout.PaneID]struct{}, len(tabCapture.panes)) | |
| 557 | 605 | for _, paneCapture := range tabCapture.panes { | |
| 558 | 606 | sealed, tail, historyErr := vt.MarshalSealedHistory(paneCapture.history) | |
| 559 | - | if historyErr != nil || paneCapture.visibleErr != nil { | |
| 560 | - | return nil, fmt.Errorf("terminal snapshot pane %s: history=%w visible=%w", paneCapture.id, historyErr, paneCapture.visibleErr) | |
| 607 | + | if historyErr != nil { | |
| 608 | + | return nil, fmt.Errorf("terminal snapshot pane %s history: %w", paneCapture.id, historyErr) | |
| 609 | + | } | |
| 610 | + | if paneCapture.visibleErr != nil { | |
| 611 | + | return nil, fmt.Errorf("terminal snapshot pane %s visible: %w", paneCapture.id, paneCapture.visibleErr) | |
| 561 | 612 | } | |
| 562 | 613 | tabSnap.Panes = append(tabSnap.Panes, snapcodec.Pane{ID: paneCapture.id, StableID: paneCapture.stableID, Cwd: paneCapture.cwd, SealedChunks: sealed, Tail: tail, Visible: paneCapture.visible, Process: paneCapture.process}) | |
| 563 | 614 | paneIDs[paneCapture.id] = struct{}{} | |
| @@ -580,16 +631,75 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) | |||
| 580 | 631 | capture.session.snapshotPendingCaptures-- | |
| 581 | 632 | } | |
| 582 | 633 | capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 | |
| 634 | + | if capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil { | |
| 635 | + | // The interval is measured from completion, including a failed | |
| 636 | + | // attempt. A forced teardown checkpoint deliberately leaves it alone. | |
| 637 | + | capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval) | |
| 638 | + | } | |
| 583 | 639 | if succeeded && capture.session.snapshotGeneration == capture.generation { | |
| 584 | 640 | capture.session.snapDirty.Store(false) | |
| 641 | + | capture.session.snapshotFailureSig = "" | |
| 585 | 642 | } else if !succeeded { | |
| 586 | 643 | capture.session.snapDirty.Store(true) | |
| 587 | 644 | } | |
| 588 | 645 | capture.session.signalSnapshotChangedLocked() | |
| 646 | + | wake := capture.session.snapshotWake | |
| 589 | 647 | capture.session.snapshotMu.Unlock() | |
| 648 | + | if wake != nil { | |
| 649 | + | select { | |
| 650 | + | case wake <- struct{}{}: | |
| 651 | + | default: | |
| 652 | + | } | |
| 653 | + | } | |
| 654 | + | if succeeded { | |
| 655 | + | d.clearSnapshotFailure() | |
| 656 | + | } | |
| 590 | 657 | }) | |
| 591 | 658 | } | |
| 592 | 659 | ||
| 660 | + | // reportSnapshotFailure records every failed persistence attempt but only | |
| 661 | + | // displays a toast when its stable phase/error class changes. This keeps a | |
| 662 | + | // blocked disk from repeatedly repainting every attached client while retaining | |
| 663 | + | // the count in notification history. | |
| 664 | + | func (d *Daemon) reportSnapshotFailure(capture *snapshotCapture, phase string, cause error) { | |
| 665 | + | if d == nil || capture == nil || capture.session == nil || cause == nil { | |
| 666 | + | return | |
| 667 | + | } | |
| 668 | + | signature := snapshotFailureSignature(phase, cause) | |
| 669 | + | capture.session.snapshotMu.Lock() | |
| 670 | + | capture.session.snapshotFailureSig = signature | |
| 671 | + | capture.session.snapshotMu.Unlock() | |
| 672 | + | ||
| 673 | + | d.snapshotNoticeMu.Lock() | |
| 674 | + | changed := d.snapshotActiveFailureSignature != signature | |
| 675 | + | d.snapshotActiveFailureSignature = signature | |
| 676 | + | d.snapshotNoticeMu.Unlock() | |
| 677 | + | ||
| 678 | + | n := domain.Notification{ | |
| 679 | + | Code: domain.NoticeSnapshotWrite, | |
| 680 | + | Severity: domain.NoticeError, | |
| 681 | + | Message: "couldn't save session state; recent state may be lost on restart", | |
| 682 | + | // The history is intentionally diagnostic only at the error-class level. | |
| 683 | + | // Raw causes can include user paths and terminal content. | |
| 684 | + | Details: signature, | |
| 685 | + | Time: d.clock.Now(), | |
| 686 | + | } | |
| 687 | + | n, _ = d.notices.recordSnapshotFailure(n) | |
| 688 | + | d.log.Warn("writing session snapshot failed", "phase", phase, "class", signature, "session", capture.name) | |
| 689 | + | if changed { | |
| 690 | + | d.deliverGlobal(n) | |
| 691 | + | } | |
| 692 | + | } | |
| 693 | + | ||
| 694 | + | func (d *Daemon) clearSnapshotFailure() { | |
| 695 | + | if d == nil { | |
| 696 | + | return | |
| 697 | + | } | |
| 698 | + | d.snapshotNoticeMu.Lock() | |
| 699 | + | d.snapshotActiveFailureSignature = "" | |
| 700 | + | d.snapshotNoticeMu.Unlock() | |
| 701 | + | } | |
| 702 | + | ||
| 593 | 703 | func (d *Daemon) startSnapshotEncodeWorker() { | |
| 594 | 704 | if d == nil { | |
| 595 | 705 | return | |
| @@ -641,12 +751,10 @@ func (d *Daemon) startSnapshotEncodeWorker() { | |||
| 641 | 751 | } | |
| 642 | 752 | d.clearSnapshotWorkerInFlight(workerID, capture) | |
| 643 | 753 | if err != nil && workerCtx.Err() == nil { | |
| 644 | - | d.log.Warn("writing session snapshot failed", "err", err, "session", capture.name) | |
| 645 | 754 | // Global, not session-scoped: by now the session may already be | |
| 646 | 755 | // torn down, so a session notice would be dead-on-arrival. No lock | |
| 647 | 756 | // is held here (clearSnapshotWorkerInFlight released its own). | |
| 648 | - | d.NotifyGlobal(domain.NoticeError, domain.NoticeSnapshotWrite, | |
| 649 | - | "couldn't save session "+capture.name+"; recent state may be lost on restart", err) | |
| 757 | + | d.reportSnapshotFailure(capture, "publish", err) | |
| 650 | 758 | } | |
| 651 | 759 | d.finishSnapshotCapture(capture, err == nil && workerCtx.Err() == nil) | |
| 652 | 760 | return workerCtx.Err() == nil | |
| @@ -815,6 +923,10 @@ func (d *Daemon) capturePaneProcess(pty interface{ ForegroundPgid() (int, error) | |||
| 815 | 923 | } | |
| 816 | 924 | ||
| 817 | 925 | func (d *Daemon) snapshotSaver(ctx context.Context) { | |
| 926 | + | if d.snapshotRepository != nil { | |
| 927 | + | d.snapshotRepositorySaver(ctx) | |
| 928 | + | return | |
| 929 | + | } | |
| 818 | 930 | checkpoint := func() { | |
| 819 | 931 | d.mu.Lock() | |
| 820 | 932 | sessions := d.sessionsSnapshotLocked() | |
| @@ -842,6 +954,71 @@ func (d *Daemon) snapshotSaver(ctx context.Context) { | |||
| 842 | 954 | } | |
| 843 | 955 | } | |
| 844 | 956 | ||
| 957 | + | // snapshotRepositorySaver waits until the earliest completion-derived routine | |
| 958 | + | // eligibility. State changes wake it immediately, so a newly dirty session is | |
| 959 | + | // captured without waiting for an unrelated session's interval. | |
| 960 | + | func (d *Daemon) snapshotRepositorySaver(ctx context.Context) { | |
| 961 | + | if d == nil { | |
| 962 | + | return | |
| 963 | + | } | |
| 964 | + | timer := d.clock.NewTimer(24 * time.Hour) | |
| 965 | + | defer timer.Stop() | |
| 966 | + | for { | |
| 967 | + | delay := d.scheduleEligibleRepositorySnapshots() | |
| 968 | + | timer.Reset(delay) | |
| 969 | + | select { | |
| 970 | + | case <-ctx.Done(): | |
| 971 | + | return | |
| 972 | + | case <-d.snapshotWake: | |
| 973 | + | case <-timer.C(): | |
| 974 | + | } | |
| 975 | + | } | |
| 976 | + | } | |
| 977 | + | ||
| 978 | + | func (d *Daemon) scheduleEligibleRepositorySnapshots() time.Duration { | |
| 979 | + | d.mu.Lock() | |
| 980 | + | sessions := d.sessionsSnapshotLocked() | |
| 981 | + | d.mu.Unlock() | |
| 982 | + | now := d.clock.Now() | |
| 983 | + | for _, sess := range sessions { | |
| 984 | + | if !sess.snapEligible.Load() || !sess.snapDirty.Load() { | |
| 985 | + | continue | |
| 986 | + | } | |
| 987 | + | sess.snapshotMu.Lock() | |
| 988 | + | due := !sess.snapshotPending && (sess.snapshotNextEligibleAt.IsZero() || !now.Before(sess.snapshotNextEligibleAt)) | |
| 989 | + | sess.snapshotMu.Unlock() | |
| 990 | + | if due { | |
| 991 | + | d.scheduleSnapshot(sess) | |
| 992 | + | } | |
| 993 | + | } | |
| 994 | + | ||
| 995 | + | // Scheduling can synchronously fail or mark a capture pending, so inspect | |
| 996 | + | // the final state to select the actual earliest timer deadline. | |
| 997 | + | var next time.Time | |
| 998 | + | for _, sess := range sessions { | |
| 999 | + | if !sess.snapEligible.Load() || !sess.snapDirty.Load() { | |
| 1000 | + | continue | |
| 1001 | + | } | |
| 1002 | + | sess.snapshotMu.Lock() | |
| 1003 | + | eligibleAt := sess.snapshotNextEligibleAt | |
| 1004 | + | pending := sess.snapshotPending | |
| 1005 | + | sess.snapshotMu.Unlock() | |
| 1006 | + | if pending || eligibleAt.IsZero() { | |
| 1007 | + | continue | |
| 1008 | + | } | |
| 1009 | + | if next.IsZero() || eligibleAt.Before(next) { | |
| 1010 | + | next = eligibleAt | |
| 1011 | + | } | |
| 1012 | + | } | |
| 1013 | + | if next.IsZero() { | |
| 1014 | + | return 24 * time.Hour | |
| 1015 | + | } | |
| 1016 | + | if delay := next.Sub(now); delay > 0 { | |
| 1017 | + | return delay | |
| 1018 | + | } | |
| 1019 | + | return 0 | |
| 1020 | + | } | |
| 1021 | + | ||
| 845 | 1022 | func pruneTreeToPanes(n *layout.Node, panes map[layout.PaneID]struct{}) bool { | |
| 846 | 1023 | if n == nil { | |
| 847 | 1024 | return false | |
| @@ -2,6 +2,7 @@ package daemon | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 5 | "fmt" | 6 | "fmt" | ||
| 6 | "math" | 7 | "math" | ||
| 7 | "sync" | 8 | "sync" | ||
| @@ -22,14 +23,25 @@ const snapshotQueueCapacity = 1 | |||||
| 22 | // (and left dirty) rather than silently dropping a persistence acknowledgement. | 23 | // (and left dirty) rather than silently dropping a persistence acknowledgement. | ||
| 23 | const snapshotFinalQueueCapacity = 32 | 24 | const snapshotFinalQueueCapacity = 32 | ||
| 24 | 25 | ||||
| 26 | + | // snapshotAttemptKind distinguishes rate-limited routine publications from | |||
| 27 | + | // mandatory terminal checkpoints. Forced attempts never consume the routine | |||
| 28 | + | // completion window. | |||
| 29 | + | type snapshotAttemptKind uint8 | |||
| 30 | + | ||||
| 31 | + | const ( | |||
| 32 | + | snapshotAttemptRoutine snapshotAttemptKind = iota | |||
| 33 | + | snapshotAttemptForced | |||
| 34 | + | ) | |||
| 35 | + | ||||
| 25 | type snapshotCapture struct { | 36 | type snapshotCapture struct { | ||
| 26 | - | session *session | 37 | + | session *session |
| 27 | - | generation uint64 | 38 | + | attemptKind snapshotAttemptKind |
| 28 | - | name string | 39 | + | generation uint64 |
| 29 | - | createdAt uint64 | 40 | + | name string |
| 30 | - | active uint16 | 41 | + | createdAt uint64 |
| 31 | - | tabs []snapshotCaptureTab | 42 | + | active uint16 |
| 32 | - | finishOnce sync.Once | 43 | + | tabs []snapshotCaptureTab |
| 44 | + | finishOnce sync.Once | |||
| 33 | } | 45 | } | ||
| 34 | 46 | ||||
| 35 | type snapshotCaptureTab struct { | 47 | type snapshotCaptureTab struct { | ||
| @@ -57,6 +69,24 @@ type snapshotCapturePane struct { | |||||
| 57 | process *snapcodec.Process | 69 | process *snapcodec.Process | ||
| 58 | } | 70 | } | ||
| 59 | 71 | ||||
| 72 | + | // snapshotFailureSignature classifies a persistence failure without retaining | |||
| 73 | + | // error text. Error strings often contain object content or filesystem paths, | |||
| 74 | + | // neither of which is stable enough for deduplication or suitable for a global | |||
| 75 | + | // notice history. | |||
| 76 | + | func snapshotFailureSignature(phase string, err error) string { | |||
| 77 | + | if err == nil { | |||
| 78 | + | return phase + ":none" | |||
| 79 | + | } | |||
| 80 | + | switch { | |||
| 81 | + | case errors.Is(err, context.Canceled): | |||
| 82 | + | return phase + ":canceled" | |||
| 83 | + | case errors.Is(err, context.DeadlineExceeded): | |||
| 84 | + | return phase + ":deadline" | |||
| 85 | + | default: | |||
| 86 | + | return fmt.Sprintf("%s:%T", phase, err) | |||
| 87 | + | } | |||
| 88 | + | } | |||
| 89 | + | ||||
| 60 | const ( | 90 | const ( | ||
| 61 | // Routine checkpoints are rate limited from completion; forced teardown | 91 | // Routine checkpoints are rate limited from completion; forced teardown | ||
| 62 | // checkpoints use the same bounded worker but never consume this window. | 92 | // checkpoints use the same bounded worker but never consume this window. | ||
| @@ -266,7 +296,7 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err | |||||
| 266 | return nil | 296 | return nil | ||
| 267 | } | 297 | } | ||
| 268 | createdAt := int64(snap.CreatedAt) | 298 | createdAt := int64(snap.CreatedAt) | ||
| 269 | - | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt} | 299 | + | sess := &session{name: snap.Name, ctx: sctx, cancel: cancel, tabs: opened, active: int(snap.Active), terminal: restoreTerm, env: restoreEnv, createdAt: createdAt, snapshotWake: d.snapshotWake} |
| 270 | sess.snapEligible.Store(true) | 300 | sess.snapEligible.Store(true) | ||
| 271 | if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 { | 301 | if len(snap.Tabs) > 0 && len(snap.Tabs[0].Panes) > 0 { | ||
| 272 | sess.cwd = snap.Tabs[0].Panes[0].Cwd | 302 | sess.cwd = snap.Tabs[0].Panes[0].Cwd | ||
| @@ -308,7 +338,14 @@ func markSnapshotDirty(sess *session) { | |||||
| 308 | sess.snapshotGeneration++ | 338 | sess.snapshotGeneration++ | ||
| 309 | sess.snapDirty.Store(true) | 339 | sess.snapDirty.Store(true) | ||
| 310 | sess.signalSnapshotChangedLocked() | 340 | sess.signalSnapshotChangedLocked() | ||
| 341 | + | wake := sess.snapshotWake | |||
| 311 | sess.snapshotMu.Unlock() | 342 | sess.snapshotMu.Unlock() | ||
| 343 | + | if wake != nil { | |||
| 344 | + | select { | |||
| 345 | + | case wake <- struct{}{}: | |||
| 346 | + | default: | |||
| 347 | + | } | |||
| 348 | + | } | |||
| 312 | } | 349 | } | ||
| 313 | 350 | ||||
| 314 | // snapshotChangeLocked returns a channel closed by the next snapshot state | 351 | // snapshotChangeLocked returns a channel closed by the next snapshot state | ||
| @@ -383,18 +420,29 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||||
| 383 | if d == nil || sess == nil || !d.snapsEnabled || (d.snapshotRepository == nil && d.snaps == nil) || !sess.snapEligible.Load() { | 420 | if d == nil || sess == nil || !d.snapsEnabled || (d.snapshotRepository == nil && d.snaps == nil) || !sess.snapEligible.Load() { | ||
| 384 | return true | 421 | return true | ||
| 385 | } | 422 | } | ||
| 423 | + | kind := snapshotAttemptRoutine | |||
| 424 | + | if final { | |||
| 425 | + | kind = snapshotAttemptForced | |||
| 426 | + | } | |||
| 386 | sess.snapshotMu.Lock() | 427 | sess.snapshotMu.Lock() | ||
| 387 | - | if !sess.snapDirty.Load() || (!final && sess.snapshotPending) { | 428 | + | routineRateLimited := d.snapshotRepository != nil |
| 429 | + | if !sess.snapDirty.Load() || (!final && (sess.snapshotPending || (routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)))) { | |||
| 388 | sess.snapshotMu.Unlock() | 430 | sess.snapshotMu.Unlock() | ||
| 389 | return true | 431 | return true | ||
| 390 | } | 432 | } | ||
| 391 | generation := sess.snapshotGeneration | 433 | generation := sess.snapshotGeneration | ||
| 434 | + | sess.snapshotCapturedGeneration = generation | |||
| 435 | + | sess.snapshotAttempted = true | |||
| 436 | + | sess.snapshotAttemptKind = kind | |||
| 392 | sess.snapshotPendingCaptures++ | 437 | sess.snapshotPendingCaptures++ | ||
| 393 | sess.snapshotPending = true | 438 | sess.snapshotPending = true | ||
| 394 | sess.signalSnapshotChangedLocked() | 439 | sess.signalSnapshotChangedLocked() | ||
| 395 | sess.snapshotMu.Unlock() | 440 | sess.snapshotMu.Unlock() | ||
| 396 | 441 | ||||
| 397 | capture, ok := d.captureSnapshotState(sess, generation) | 442 | capture, ok := d.captureSnapshotState(sess, generation) | ||
| 443 | + | if capture != nil { | |||
| 444 | + | capture.attemptKind = kind | |||
| 445 | + | } | |||
| 398 | if !ok { | 446 | if !ok { | ||
| 399 | d.finishSnapshotCapture(capture, false) | 447 | d.finishSnapshotCapture(capture, false) | ||
| 400 | return false | 448 | return false | ||
| @@ -556,8 +604,11 @@ func (d *Daemon) encodeSnapshotCapture(capture *snapshotCapture) ([]byte, error) | |||||
| 556 | paneIDs := make(map[layout.PaneID]struct{}, len(tabCapture.panes)) | 604 | paneIDs := make(map[layout.PaneID]struct{}, len(tabCapture.panes)) | ||
| 557 | for _, paneCapture := range tabCapture.panes { | 605 | for _, paneCapture := range tabCapture.panes { | ||
| 558 | sealed, tail, historyErr := vt.MarshalSealedHistory(paneCapture.history) | 606 | sealed, tail, historyErr := vt.MarshalSealedHistory(paneCapture.history) | ||
| 559 | - | if historyErr != nil || paneCapture.visibleErr != nil { | 607 | + | if historyErr != nil { |
| 560 | - | return nil, fmt.Errorf("terminal snapshot pane %s: history=%w visible=%w", paneCapture.id, historyErr, paneCapture.visibleErr) | 608 | + | return nil, fmt.Errorf("terminal snapshot pane %s history: %w", paneCapture.id, historyErr) |
| 609 | + | } | |||
| 610 | + | if paneCapture.visibleErr != nil { | |||
| 611 | + | return nil, fmt.Errorf("terminal snapshot pane %s visible: %w", paneCapture.id, paneCapture.visibleErr) | |||
| 561 | } | 612 | } | ||
| 562 | tabSnap.Panes = append(tabSnap.Panes, snapcodec.Pane{ID: paneCapture.id, StableID: paneCapture.stableID, Cwd: paneCapture.cwd, SealedChunks: sealed, Tail: tail, Visible: paneCapture.visible, Process: paneCapture.process}) | 613 | tabSnap.Panes = append(tabSnap.Panes, snapcodec.Pane{ID: paneCapture.id, StableID: paneCapture.stableID, Cwd: paneCapture.cwd, SealedChunks: sealed, Tail: tail, Visible: paneCapture.visible, Process: paneCapture.process}) | ||
| 563 | paneIDs[paneCapture.id] = struct{}{} | 614 | paneIDs[paneCapture.id] = struct{}{} | ||
| @@ -580,16 +631,75 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) | |||||
| 580 | capture.session.snapshotPendingCaptures-- | 631 | capture.session.snapshotPendingCaptures-- | ||
| 581 | } | 632 | } | ||
| 582 | capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 | 633 | capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 | ||
| 634 | + | if capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil { | |||
| 635 | + | // The interval is measured from completion, including a failed | |||
| 636 | + | // attempt. A forced teardown checkpoint deliberately leaves it alone. | |||
| 637 | + | capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval) | |||
| 638 | + | } | |||
| 583 | if succeeded && capture.session.snapshotGeneration == capture.generation { | 639 | if succeeded && capture.session.snapshotGeneration == capture.generation { | ||
| 584 | capture.session.snapDirty.Store(false) | 640 | capture.session.snapDirty.Store(false) | ||
| 641 | + | capture.session.snapshotFailureSig = "" | |||
| 585 | } else if !succeeded { | 642 | } else if !succeeded { | ||
| 586 | capture.session.snapDirty.Store(true) | 643 | capture.session.snapDirty.Store(true) | ||
| 587 | } | 644 | } | ||
| 588 | capture.session.signalSnapshotChangedLocked() | 645 | capture.session.signalSnapshotChangedLocked() | ||
| 646 | + | wake := capture.session.snapshotWake | |||
| 589 | capture.session.snapshotMu.Unlock() | 647 | capture.session.snapshotMu.Unlock() | ||
| 648 | + | if wake != nil { | |||
| 649 | + | select { | |||
| 650 | + | case wake <- struct{}{}: | |||
| 651 | + | default: | |||
| 652 | + | } | |||
| 653 | + | } | |||
| 654 | + | if succeeded { | |||
| 655 | + | d.clearSnapshotFailure() | |||
| 656 | + | } | |||
| 590 | }) | 657 | }) | ||
| 591 | } | 658 | } | ||
| 592 | 659 | ||||
| 660 | + | // reportSnapshotFailure records every failed persistence attempt but only | |||
| 661 | + | // displays a toast when its stable phase/error class changes. This keeps a | |||
| 662 | + | // blocked disk from repeatedly repainting every attached client while retaining | |||
| 663 | + | // the count in notification history. | |||
| 664 | + | func (d *Daemon) reportSnapshotFailure(capture *snapshotCapture, phase string, cause error) { | |||
| 665 | + | if d == nil || capture == nil || capture.session == nil || cause == nil { | |||
| 666 | + | return | |||
| 667 | + | } | |||
| 668 | + | signature := snapshotFailureSignature(phase, cause) | |||
| 669 | + | capture.session.snapshotMu.Lock() | |||
| 670 | + | capture.session.snapshotFailureSig = signature | |||
| 671 | + | capture.session.snapshotMu.Unlock() | |||
| 672 | + | ||||
| 673 | + | d.snapshotNoticeMu.Lock() | |||
| 674 | + | changed := d.snapshotActiveFailureSignature != signature | |||
| 675 | + | d.snapshotActiveFailureSignature = signature | |||
| 676 | + | d.snapshotNoticeMu.Unlock() | |||
| 677 | + | ||||
| 678 | + | n := domain.Notification{ | |||
| 679 | + | Code: domain.NoticeSnapshotWrite, | |||
| 680 | + | Severity: domain.NoticeError, | |||
| 681 | + | Message: "couldn't save session state; recent state may be lost on restart", | |||
| 682 | + | // The history is intentionally diagnostic only at the error-class level. | |||
| 683 | + | // Raw causes can include user paths and terminal content. | |||
| 684 | + | Details: signature, | |||
| 685 | + | Time: d.clock.Now(), | |||
| 686 | + | } | |||
| 687 | + | n, _ = d.notices.recordSnapshotFailure(n) | |||
| 688 | + | d.log.Warn("writing session snapshot failed", "phase", phase, "class", signature, "session", capture.name) | |||
| 689 | + | if changed { | |||
| 690 | + | d.deliverGlobal(n) | |||
| 691 | + | } | |||
| 692 | + | } | |||
| 693 | + | ||||
| 694 | + | func (d *Daemon) clearSnapshotFailure() { | |||
| 695 | + | if d == nil { | |||
| 696 | + | return | |||
| 697 | + | } | |||
| 698 | + | d.snapshotNoticeMu.Lock() | |||
| 699 | + | d.snapshotActiveFailureSignature = "" | |||
| 700 | + | d.snapshotNoticeMu.Unlock() | |||
| 701 | + | } | |||
| 702 | + | ||||
| 593 | func (d *Daemon) startSnapshotEncodeWorker() { | 703 | func (d *Daemon) startSnapshotEncodeWorker() { | ||
| 594 | if d == nil { | 704 | if d == nil { | ||
| 595 | return | 705 | return | ||
| @@ -641,12 +751,10 @@ func (d *Daemon) startSnapshotEncodeWorker() { | |||||
| 641 | } | 751 | } | ||
| 642 | d.clearSnapshotWorkerInFlight(workerID, capture) | 752 | d.clearSnapshotWorkerInFlight(workerID, capture) | ||
| 643 | if err != nil && workerCtx.Err() == nil { | 753 | if err != nil && workerCtx.Err() == nil { | ||
| 644 | - | d.log.Warn("writing session snapshot failed", "err", err, "session", capture.name) | |||
| 645 | // Global, not session-scoped: by now the session may already be | 754 | // Global, not session-scoped: by now the session may already be | ||
| 646 | // torn down, so a session notice would be dead-on-arrival. No lock | 755 | // torn down, so a session notice would be dead-on-arrival. No lock | ||
| 647 | // is held here (clearSnapshotWorkerInFlight released its own). | 756 | // is held here (clearSnapshotWorkerInFlight released its own). | ||
| 648 | - | d.NotifyGlobal(domain.NoticeError, domain.NoticeSnapshotWrite, | 757 | + | d.reportSnapshotFailure(capture, "publish", err) |
| 649 | - | "couldn't save session "+capture.name+"; recent state may be lost on restart", err) | |||
| 650 | } | 758 | } | ||
| 651 | d.finishSnapshotCapture(capture, err == nil && workerCtx.Err() == nil) | 759 | d.finishSnapshotCapture(capture, err == nil && workerCtx.Err() == nil) | ||
| 652 | return workerCtx.Err() == nil | 760 | return workerCtx.Err() == nil | ||
| @@ -815,6 +923,10 @@ func (d *Daemon) capturePaneProcess(pty interface{ ForegroundPgid() (int, error) | |||||
| 815 | } | 923 | } | ||
| 816 | 924 | ||||
| 817 | func (d *Daemon) snapshotSaver(ctx context.Context) { | 925 | func (d *Daemon) snapshotSaver(ctx context.Context) { | ||
| 926 | + | if d.snapshotRepository != nil { | |||
| 927 | + | d.snapshotRepositorySaver(ctx) | |||
| 928 | + | return | |||
| 929 | + | } | |||
| 818 | checkpoint := func() { | 930 | checkpoint := func() { | ||
| 819 | d.mu.Lock() | 931 | d.mu.Lock() | ||
| 820 | sessions := d.sessionsSnapshotLocked() | 932 | sessions := d.sessionsSnapshotLocked() | ||
| @@ -842,6 +954,71 @@ func (d *Daemon) snapshotSaver(ctx context.Context) { | |||||
| 842 | } | 954 | } | ||
| 843 | } | 955 | } | ||
| 844 | 956 | ||||
| 957 | + | // snapshotRepositorySaver waits until the earliest completion-derived routine | |||
| 958 | + | // eligibility. State changes wake it immediately, so a newly dirty session is | |||
| 959 | + | // captured without waiting for an unrelated session's interval. | |||
| 960 | + | func (d *Daemon) snapshotRepositorySaver(ctx context.Context) { | |||
| 961 | + | if d == nil { | |||
| 962 | + | return | |||
| 963 | + | } | |||
| 964 | + | timer := d.clock.NewTimer(24 * time.Hour) | |||
| 965 | + | defer timer.Stop() | |||
| 966 | + | for { | |||
| 967 | + | delay := d.scheduleEligibleRepositorySnapshots() | |||
| 968 | + | timer.Reset(delay) | |||
| 969 | + | select { | |||
| 970 | + | case <-ctx.Done(): | |||
| 971 | + | return | |||
| 972 | + | case <-d.snapshotWake: | |||
| 973 | + | case <-timer.C(): | |||
| 974 | + | } | |||
| 975 | + | } | |||
| 976 | + | } | |||
| 977 | + | ||||
| 978 | + | func (d *Daemon) scheduleEligibleRepositorySnapshots() time.Duration { | |||
| 979 | + | d.mu.Lock() | |||
| 980 | + | sessions := d.sessionsSnapshotLocked() | |||
| 981 | + | d.mu.Unlock() | |||
| 982 | + | now := d.clock.Now() | |||
| 983 | + | for _, sess := range sessions { | |||
| 984 | + | if !sess.snapEligible.Load() || !sess.snapDirty.Load() { | |||
| 985 | + | continue | |||
| 986 | + | } | |||
| 987 | + | sess.snapshotMu.Lock() | |||
| 988 | + | due := !sess.snapshotPending && (sess.snapshotNextEligibleAt.IsZero() || !now.Before(sess.snapshotNextEligibleAt)) | |||
| 989 | + | sess.snapshotMu.Unlock() | |||
| 990 | + | if due { | |||
| 991 | + | d.scheduleSnapshot(sess) | |||
| 992 | + | } | |||
| 993 | + | } | |||
| 994 | + | ||||
| 995 | + | // Scheduling can synchronously fail or mark a capture pending, so inspect | |||
| 996 | + | // the final state to select the actual earliest timer deadline. | |||
| 997 | + | var next time.Time | |||
| 998 | + | for _, sess := range sessions { | |||
| 999 | + | if !sess.snapEligible.Load() || !sess.snapDirty.Load() { | |||
| 1000 | + | continue | |||
| 1001 | + | } | |||
| 1002 | + | sess.snapshotMu.Lock() | |||
| 1003 | + | eligibleAt := sess.snapshotNextEligibleAt | |||
| 1004 | + | pending := sess.snapshotPending | |||
| 1005 | + | sess.snapshotMu.Unlock() | |||
| 1006 | + | if pending || eligibleAt.IsZero() { | |||
| 1007 | + | continue | |||
| 1008 | + | } | |||
| 1009 | + | if next.IsZero() || eligibleAt.Before(next) { | |||
| 1010 | + | next = eligibleAt | |||
| 1011 | + | } | |||
| 1012 | + | } | |||
| 1013 | + | if next.IsZero() { | |||
| 1014 | + | return 24 * time.Hour | |||
| 1015 | + | } | |||
| 1016 | + | if delay := next.Sub(now); delay > 0 { | |||
| 1017 | + | return delay | |||
| 1018 | + | } | |||
| 1019 | + | return 0 | |||
| 1020 | + | } | |||
| 1021 | + | ||||
| 845 | func pruneTreeToPanes(n *layout.Node, panes map[layout.PaneID]struct{}) bool { | 1022 | func pruneTreeToPanes(n *layout.Node, panes map[layout.PaneID]struct{}) bool { | ||
| 846 | if n == nil { | 1023 | if n == nil { | ||
| 847 | return false | 1024 | return false | ||
A internal/usecase/daemon/snapshot_failure_notice_test.go +27 -0
| @@ -0,0 +1,27 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "testing" | |
| 6 | + | ||
| 7 | + | "github.com/stretchr/testify/require" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | func TestSnapshotFailureNoticeDeduplicatesStableCauseAndClearsOnSuccess(t *testing.T) { | |
| 11 | + | d, _, client, _ := newNoticeFixture(t, newNoticeClock()) | |
| 12 | + | capture := &snapshotCapture{session: newSnapshotTestSession(t, "failed", false, "/work"), name: "failed"} | |
| 13 | + | ||
| 14 | + | d.reportSnapshotFailure(capture, "publish", errors.New("first transient failure")) | |
| 15 | + | d.reportSnapshotFailure(capture, "publish", errors.New("same class, changed details")) | |
| 16 | + | ||
| 17 | + | history := d.notices.history() | |
| 18 | + | require.Len(t, history, 1) | |
| 19 | + | require.EqualValues(t, 2, history[0].Count) | |
| 20 | + | require.Len(t, awaitToastCount(t, client, 1), 1) | |
| 21 | + | ||
| 22 | + | d.finishSnapshotCapture(capture, true) | |
| 23 | + | d.reportSnapshotFailure(capture, "publish", errors.New("failure after success")) | |
| 24 | + | require.Len(t, awaitToastCount(t, client, 1), 1) | |
| 25 | + | require.Len(t, d.notices.history(), 1) | |
| 26 | + | require.EqualValues(t, 3, d.notices.history()[0].Count) | |
| 27 | + | } | |
| @@ -0,0 +1,27 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "testing" | |||
| 6 | + | ||||
| 7 | + | "github.com/stretchr/testify/require" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | func TestSnapshotFailureNoticeDeduplicatesStableCauseAndClearsOnSuccess(t *testing.T) { | |||
| 11 | + | d, _, client, _ := newNoticeFixture(t, newNoticeClock()) | |||
| 12 | + | capture := &snapshotCapture{session: newSnapshotTestSession(t, "failed", false, "/work"), name: "failed"} | |||
| 13 | + | ||||
| 14 | + | d.reportSnapshotFailure(capture, "publish", errors.New("first transient failure")) | |||
| 15 | + | d.reportSnapshotFailure(capture, "publish", errors.New("same class, changed details")) | |||
| 16 | + | ||||
| 17 | + | history := d.notices.history() | |||
| 18 | + | require.Len(t, history, 1) | |||
| 19 | + | require.EqualValues(t, 2, history[0].Count) | |||
| 20 | + | require.Len(t, awaitToastCount(t, client, 1), 1) | |||
| 21 | + | ||||
| 22 | + | d.finishSnapshotCapture(capture, true) | |||
| 23 | + | d.reportSnapshotFailure(capture, "publish", errors.New("failure after success")) | |||
| 24 | + | require.Len(t, awaitToastCount(t, client, 1), 1) | |||
| 25 | + | require.Len(t, d.notices.history(), 1) | |||
| 26 | + | require.EqualValues(t, 3, d.notices.history()[0].Count) | |||
| 27 | + | } | |||
A internal/usecase/daemon/snapshot_failure_test.go +22 -0
| @@ -0,0 +1,22 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "os" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/stretchr/testify/require" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | func TestSnapshotFailureSignatureIsStableAndNilSafe(t *testing.T) { | |
| 12 | + | t.Parallel() | |
| 13 | + | ||
| 14 | + | first := &os.PathError{Op: "write", Path: "/private/first", Err: os.ErrPermission} | |
| 15 | + | second := &os.PathError{Op: "write", Path: "/volatile/second", Err: os.ErrPermission} | |
| 16 | + | ||
| 17 | + | require.Equal(t, snapshotFailureSignature("publish", first), snapshotFailureSignature("publish", second)) | |
| 18 | + | require.NotContains(t, snapshotFailureSignature("publish", first), "/private/first") | |
| 19 | + | require.NotContains(t, snapshotFailureSignature("publish", nil), "%!w(<nil>)") | |
| 20 | + | require.NotEqual(t, snapshotFailureSignature("publish", first), snapshotFailureSignature("encode", first)) | |
| 21 | + | require.NotEqual(t, snapshotFailureSignature("publish", first), snapshotFailureSignature("publish", errors.New("other"))) | |
| 22 | + | } | |
| @@ -0,0 +1,22 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "os" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/stretchr/testify/require" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | func TestSnapshotFailureSignatureIsStableAndNilSafe(t *testing.T) { | |||
| 12 | + | t.Parallel() | |||
| 13 | + | ||||
| 14 | + | first := &os.PathError{Op: "write", Path: "/private/first", Err: os.ErrPermission} | |||
| 15 | + | second := &os.PathError{Op: "write", Path: "/volatile/second", Err: os.ErrPermission} | |||
| 16 | + | ||||
| 17 | + | require.Equal(t, snapshotFailureSignature("publish", first), snapshotFailureSignature("publish", second)) | |||
| 18 | + | require.NotContains(t, snapshotFailureSignature("publish", first), "/private/first") | |||
| 19 | + | require.NotContains(t, snapshotFailureSignature("publish", nil), "%!w(<nil>)") | |||
| 20 | + | require.NotEqual(t, snapshotFailureSignature("publish", first), snapshotFailureSignature("encode", first)) | |||
| 21 | + | require.NotEqual(t, snapshotFailureSignature("publish", first), snapshotFailureSignature("publish", errors.New("other"))) | |||
| 22 | + | } | |||
M internal/usecase/daemon/snapshot_notice_test.go +2 -2
| @@ -40,11 +40,11 @@ func TestSnapshotWriteFailureNotifiesGlobally(t *testing.T) { | |||
| 40 | 40 | // poll-only completion signal. No require runs on the worker goroutine. | |
| 41 | 41 | toasts := awaitToastCount(t, survivorClient, 1) | |
| 42 | 42 | require.Equal(t, domain.NoticeSnapshotWrite, toasts[0].Code) | |
| 43 | - | require.Equal(t, "couldn't save session failing; recent state may be lost on restart", toasts[0].Message) | |
| 43 | + | require.Equal(t, "couldn't save session state; recent state may be lost on restart", toasts[0].Message) | |
| 44 | 44 | require.Equal(t, domain.NoticeError, toasts[0].Severity) | |
| 45 | 45 | require.Equal(t, domain.SessionID(""), toasts[0].SessionID, | |
| 46 | 46 | "snapshot write failures are global, never scoped to the (possibly gone) failing session") | |
| 47 | - | require.Contains(t, toasts[0].Details, "disk full") | |
| 47 | + | require.Equal(t, "publish:*errors.errorString", toasts[0].Details) | |
| 48 | 48 | } | |
| 49 | 49 | ||
| 50 | 50 | // TestCloseLastTabSaturatedSnapshotSurfacesNotice proves that when closing the | |
| @@ -40,11 +40,11 @@ func TestSnapshotWriteFailureNotifiesGlobally(t *testing.T) { | |||||
| 40 | // poll-only completion signal. No require runs on the worker goroutine. | 40 | // poll-only completion signal. No require runs on the worker goroutine. | ||
| 41 | toasts := awaitToastCount(t, survivorClient, 1) | 41 | toasts := awaitToastCount(t, survivorClient, 1) | ||
| 42 | require.Equal(t, domain.NoticeSnapshotWrite, toasts[0].Code) | 42 | require.Equal(t, domain.NoticeSnapshotWrite, toasts[0].Code) | ||
| 43 | - | require.Equal(t, "couldn't save session failing; recent state may be lost on restart", toasts[0].Message) | 43 | + | require.Equal(t, "couldn't save session state; recent state may be lost on restart", toasts[0].Message) |
| 44 | require.Equal(t, domain.NoticeError, toasts[0].Severity) | 44 | require.Equal(t, domain.NoticeError, toasts[0].Severity) | ||
| 45 | require.Equal(t, domain.SessionID(""), toasts[0].SessionID, | 45 | require.Equal(t, domain.SessionID(""), toasts[0].SessionID, | ||
| 46 | "snapshot write failures are global, never scoped to the (possibly gone) failing session") | 46 | "snapshot write failures are global, never scoped to the (possibly gone) failing session") | ||
| 47 | - | require.Contains(t, toasts[0].Details, "disk full") | 47 | + | require.Equal(t, "publish:*errors.errorString", toasts[0].Details) |
| 48 | } | 48 | } | ||
| 49 | 49 | ||||
| 50 | // TestCloseLastTabSaturatedSnapshotSurfacesNotice proves that when closing the | 50 | // TestCloseLastTabSaturatedSnapshotSurfacesNotice proves that when closing the | ||
A internal/usecase/daemon/snapshot_scheduler_test.go +62 -0
| @@ -0,0 +1,62 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "sync" | |
| 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 TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) { | |
| 18 | + | clock := &snapshotSchedulerClock{now: time.Unix(100, 0)} | |
| 19 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |
| 20 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |
| 21 | + | WithSnapshotRepository(repository, nil)(d) | |
| 22 | + | startSnapshotEncodeWorker(t, d) | |
| 23 | + | ||
| 24 | + | var publishes atomic.Int32 | |
| 25 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ ports.SnapshotPublication) error { | |
| 26 | + | publishes.Add(1) | |
| 27 | + | return nil | |
| 28 | + | }).Times(2) | |
| 29 | + | ||
| 30 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |
| 31 | + | markSnapshotDirty(sess) | |
| 32 | + | require.True(t, d.scheduleSnapshot(sess)) | |
| 33 | + | awaitSnapshotClean(t, sess) | |
| 34 | + | ||
| 35 | + | sess.snapshotMu.Lock() | |
| 36 | + | eligibleAt := sess.snapshotNextEligibleAt | |
| 37 | + | sess.snapshotMu.Unlock() | |
| 38 | + | require.Equal(t, time.Unix(100, 0).Add(snapshotInterval), eligibleAt) | |
| 39 | + | ||
| 40 | + | markSnapshotDirty(sess) | |
| 41 | + | require.True(t, d.scheduleSnapshot(sess)) | |
| 42 | + | require.Eventually(t, func() bool { return publishes.Load() == 1 }, time.Second, time.Millisecond) | |
| 43 | + | ||
| 44 | + | require.True(t, d.scheduleFinalSnapshot(sess)) | |
| 45 | + | awaitSnapshotClean(t, sess) | |
| 46 | + | sess.snapshotMu.Lock() | |
| 47 | + | require.Equal(t, eligibleAt, sess.snapshotNextEligibleAt) | |
| 48 | + | sess.snapshotMu.Unlock() | |
| 49 | + | } | |
| 50 | + | ||
| 51 | + | type snapshotSchedulerClock struct { | |
| 52 | + | mu sync.Mutex | |
| 53 | + | now time.Time | |
| 54 | + | } | |
| 55 | + | ||
| 56 | + | func (c *snapshotSchedulerClock) Now() time.Time { | |
| 57 | + | c.mu.Lock() | |
| 58 | + | defer c.mu.Unlock() | |
| 59 | + | return c.now | |
| 60 | + | } | |
| 61 | + | ||
| 62 | + | func (*snapshotSchedulerClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} } | |
| @@ -0,0 +1,62 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "sync" | |||
| 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 TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) { | |||
| 18 | + | clock := &snapshotSchedulerClock{now: time.Unix(100, 0)} | |||
| 19 | + | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock) | |||
| 20 | + | repository := portsmocks.NewMockSnapshotRepository(t) | |||
| 21 | + | WithSnapshotRepository(repository, nil)(d) | |||
| 22 | + | startSnapshotEncodeWorker(t, d) | |||
| 23 | + | ||||
| 24 | + | var publishes atomic.Int32 | |||
| 25 | + | repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ ports.SnapshotPublication) error { | |||
| 26 | + | publishes.Add(1) | |||
| 27 | + | return nil | |||
| 28 | + | }).Times(2) | |||
| 29 | + | ||||
| 30 | + | sess := newSnapshotTestSession(t, "work", false, "/work") | |||
| 31 | + | markSnapshotDirty(sess) | |||
| 32 | + | require.True(t, d.scheduleSnapshot(sess)) | |||
| 33 | + | awaitSnapshotClean(t, sess) | |||
| 34 | + | ||||
| 35 | + | sess.snapshotMu.Lock() | |||
| 36 | + | eligibleAt := sess.snapshotNextEligibleAt | |||
| 37 | + | sess.snapshotMu.Unlock() | |||
| 38 | + | require.Equal(t, time.Unix(100, 0).Add(snapshotInterval), eligibleAt) | |||
| 39 | + | ||||
| 40 | + | markSnapshotDirty(sess) | |||
| 41 | + | require.True(t, d.scheduleSnapshot(sess)) | |||
| 42 | + | require.Eventually(t, func() bool { return publishes.Load() == 1 }, time.Second, time.Millisecond) | |||
| 43 | + | ||||
| 44 | + | require.True(t, d.scheduleFinalSnapshot(sess)) | |||
| 45 | + | awaitSnapshotClean(t, sess) | |||
| 46 | + | sess.snapshotMu.Lock() | |||
| 47 | + | require.Equal(t, eligibleAt, sess.snapshotNextEligibleAt) | |||
| 48 | + | sess.snapshotMu.Unlock() | |||
| 49 | + | } | |||
| 50 | + | ||||
| 51 | + | type snapshotSchedulerClock struct { | |||
| 52 | + | mu sync.Mutex | |||
| 53 | + | now time.Time | |||
| 54 | + | } | |||
| 55 | + | ||||
| 56 | + | func (c *snapshotSchedulerClock) Now() time.Time { | |||
| 57 | + | c.mu.Lock() | |||
| 58 | + | defer c.mu.Unlock() | |||
| 59 | + | return c.now | |||
| 60 | + | } | |||
| 61 | + | ||||
| 62 | + | func (*snapshotSchedulerClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} } | |||