ffeecb7 fix(daemon): enforce one snapshot shutdown deadline · vev

M internal/usecase/daemon/daemon.go +55 -25


@@ -519,7 +519,15 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
519	519		// force-close. Finally drain the conn handlers, run one defensive sweep in
520	520		// case any ordering ever leaves a session behind, and join the session
521	521		// goroutines (readers unblock via pty.Close, coordinators via ctx cancel).
522		-	checkpointIncomplete := d.shutdownAll(ports.ReasonServerShutdown)
	522	+	// One deadline owns every terminal checkpoint wait and the final worker
	523	+	// join. A repository is allowed to ignore cancellation, so spending a fresh
	524	+	// interval at either stage would make shutdown exceed its documented bound.
	525	+	var snapshotDeadline *snapshotShutdownDeadline
	526	+	if d.snapsEnabled {
	527	+	snapshotDeadline = newSnapshotShutdownDeadline(d.clock)
	528	+	defer snapshotDeadline.stop()
	529	+	}
	530	+	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
523	531		d.waitNotifies()
524	532		d.hardCancel()
525	533		d.serveCancel()

@@ -529,13 +537,9 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
529	537		d.connWg.Wait()
530	538		d.attachmentCleanupWg.Wait()
531	539		// Forced terminal checkpoints run before shutdown stops producers or the
532		-	// worker. If one consumed its bounded deadline, do not spend a second
533		-	// deadline trying to flush a repository call that already ignored it.
534		-	if checkpointIncomplete {
535		-	d.abandonSnapshotEncodeWorker()
536		-	} else {
537		-	d.stopSnapshotEncodeWorker()
538		-	}
	540	+	// worker. The same deadline bounds the worker's final drain, including a
	541	+	// repository call that ignores cancellation.
	542	+	d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
539	543		d.shutdownAll(ports.ReasonServerShutdown)
540	544		d.sessWg.Wait()
541	545		d.waitNotifies()

@@ -550,6 +554,10 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
550	554		// inserted after the snapshot: route rejects once closing is set, and both run
551	555		// under d.mu. killSession (which relocks) runs after the lock is released.
552	556		func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
	557	+	return d.shutdownAllWithSnapshotDeadline(reason, nil)
	558	+	}
	559	+
	560	+	func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapshotShutdownDeadline) (checkpointIncomplete bool) {
553	561		d.mu.Lock()
554	562		d.closing = true
555	563		for token, parked := range d.parked {

@@ -564,7 +572,7 @@ func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
564	572		return false
565	573		}
566	574		for _, s := range snapshot {
567		-	if err := d.killSession(s, reason, false); err != nil {
	575	+	if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {
568	576		checkpointIncomplete = true
569	577		d.log.Error("closing session with unpersisted terminal state", "err", err)
570	578		}

@@ -572,24 +580,46 @@ func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
572	580		return checkpointIncomplete
573	581		}
574	582
575		-	// abandonSnapshotEncodeWorker stops accepting captures and detaches a worker
576		-	// whose repository call already exceeded the terminal checkpoint deadline. A
577		-	// repository implementation is allowed to ignore context cancellation, so
578		-	// Serve cannot join that call without violating its shutdown bound.
579		-	func (d *Daemon) abandonSnapshotEncodeWorker() {
580		-	if d == nil {
581		-	return
	583	+	// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
	584	+	// checkpoints and the final snapshot worker join. Its done channel is never
	585	+	// closed by callers, so a detached worker can safely retain session state until
	586	+	// an uncooperative repository call returns.
	587	+	type snapshotShutdownDeadline struct {
	588	+	done chan struct{}
	589	+	stopCh chan struct{}
	590	+	finished chan struct{}
	591	+	stopOnce sync.Once
	592	+	}
	593	+
	594	+	func newSnapshotShutdownDeadline(clock ports.Clock) *snapshotShutdownDeadline {
	595	+	deadline := &snapshotShutdownDeadline{done: make(chan struct{}), stopCh: make(chan struct{}), finished: make(chan struct{})}
	596	+	timer := clock.NewTimer(snapshotFinalFlushTimeout)
	597	+	go func() {
	598	+	defer close(deadline.finished)
	599	+	select {
	600	+	case <-timer.C():
	601	+	close(deadline.done)
	602	+	case <-deadline.stopCh:
	603	+	timer.Stop()
	604	+	}
	605	+	}()
	606	+	return deadline
	607	+	}
	608	+
	609	+	func (d *snapshotShutdownDeadline) stop() {
	610	+	if d != nil {
	611	+	d.stopOnce.Do(func() {
	612	+	close(d.stopCh)
	613	+	<-d.finished
	614	+	})
582	615		}
583		-	d.snapshotWorkerMu.Lock()
584		-	cancel := d.snapshotWorkerCancel
585		-	if cancel == nil \|\| d.snapshotWorkerClosing {
586		-	d.snapshotWorkerMu.Unlock()
587		-	return
	616	+	}
	617	+
	618	+	func (d *snapshotShutdownDeadline) Done() <-chan struct{} {
	619	+	if d == nil {
	620	+	return nil
588	621		}
589		-	d.snapshotWorkerClosing = true
590		-	cancel()
591		-	d.snapshotWorkerMu.Unlock()
592		-	d.finishStoppedSnapshotWorker(true)
	622	+	return d.done
593	623		}
594	624
595	625		func (d Daemon) sessionsSnapshotLocked() []session {


@@ -519,7 +519,15 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
519		// force-close. Finally drain the conn handlers, run one defensive sweep in	519		// force-close. Finally drain the conn handlers, run one defensive sweep in
520		// case any ordering ever leaves a session behind, and join the session	520		// case any ordering ever leaves a session behind, and join the session
521		// goroutines (readers unblock via pty.Close, coordinators via ctx cancel).	521		// goroutines (readers unblock via pty.Close, coordinators via ctx cancel).
522	-	checkpointIncomplete := d.shutdownAll(ports.ReasonServerShutdown)	522	+	// One deadline owns every terminal checkpoint wait and the final worker
			523	+	// join. A repository is allowed to ignore cancellation, so spending a fresh
			524	+	// interval at either stage would make shutdown exceed its documented bound.
			525	+	var snapshotDeadline *snapshotShutdownDeadline
			526	+	if d.snapsEnabled {
			527	+	snapshotDeadline = newSnapshotShutdownDeadline(d.clock)
			528	+	defer snapshotDeadline.stop()
			529	+	}
			530	+	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
523		d.waitNotifies()	531		d.waitNotifies()
524		d.hardCancel()	532		d.hardCancel()
525		d.serveCancel()	533		d.serveCancel()

@@ -529,13 +537,9 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
529		d.connWg.Wait()	537		d.connWg.Wait()
530		d.attachmentCleanupWg.Wait()	538		d.attachmentCleanupWg.Wait()
531		// Forced terminal checkpoints run before shutdown stops producers or the	539		// Forced terminal checkpoints run before shutdown stops producers or the
532	-	// worker. If one consumed its bounded deadline, do not spend a second	540	+	// worker. The same deadline bounds the worker's final drain, including a
533	-	// deadline trying to flush a repository call that already ignored it.	541	+	// repository call that ignores cancellation.
534	-	if checkpointIncomplete {	542	+	d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
535	-	d.abandonSnapshotEncodeWorker()
536	-	} else {
537	-	d.stopSnapshotEncodeWorker()
538	-	}
539		d.shutdownAll(ports.ReasonServerShutdown)	543		d.shutdownAll(ports.ReasonServerShutdown)
540		d.sessWg.Wait()	544		d.sessWg.Wait()
541		d.waitNotifies()	545		d.waitNotifies()

@@ -550,6 +554,10 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
550		// inserted after the snapshot: route rejects once closing is set, and both run	554		// inserted after the snapshot: route rejects once closing is set, and both run
551		// under d.mu. killSession (which relocks) runs after the lock is released.	555		// under d.mu. killSession (which relocks) runs after the lock is released.
552		func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {	556		func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
			557	+	return d.shutdownAllWithSnapshotDeadline(reason, nil)
			558	+	}
			559	+
			560	+	func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapshotShutdownDeadline) (checkpointIncomplete bool) {
553		d.mu.Lock()	561		d.mu.Lock()
554		d.closing = true	562		d.closing = true
555		for token, parked := range d.parked {	563		for token, parked := range d.parked {

@@ -564,7 +572,7 @@ func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
564		return false	572		return false
565		}	573		}
566		for _, s := range snapshot {	574		for _, s := range snapshot {
567	-	if err := d.killSession(s, reason, false); err != nil {	575	+	if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {
568		checkpointIncomplete = true	576		checkpointIncomplete = true
569		d.log.Error("closing session with unpersisted terminal state", "err", err)	577		d.log.Error("closing session with unpersisted terminal state", "err", err)
570		}	578		}

@@ -572,24 +580,46 @@ func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
572		return checkpointIncomplete	580		return checkpointIncomplete
573		}	581		}
574			582
575	-	// abandonSnapshotEncodeWorker stops accepting captures and detaches a worker	583	+	// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
576	-	// whose repository call already exceeded the terminal checkpoint deadline. A	584	+	// checkpoints and the final snapshot worker join. Its done channel is never
577	-	// repository implementation is allowed to ignore context cancellation, so	585	+	// closed by callers, so a detached worker can safely retain session state until
578	-	// Serve cannot join that call without violating its shutdown bound.	586	+	// an uncooperative repository call returns.
579	-	func (d *Daemon) abandonSnapshotEncodeWorker() {	587	+	type snapshotShutdownDeadline struct {
580	-	if d == nil {	588	+	done chan struct{}
581	-	return	589	+	stopCh chan struct{}
			590	+	finished chan struct{}
			591	+	stopOnce sync.Once
			592	+	}
			593	+
			594	+	func newSnapshotShutdownDeadline(clock ports.Clock) *snapshotShutdownDeadline {
			595	+	deadline := &snapshotShutdownDeadline{done: make(chan struct{}), stopCh: make(chan struct{}), finished: make(chan struct{})}
			596	+	timer := clock.NewTimer(snapshotFinalFlushTimeout)
			597	+	go func() {
			598	+	defer close(deadline.finished)
			599	+	select {
			600	+	case <-timer.C():
			601	+	close(deadline.done)
			602	+	case <-deadline.stopCh:
			603	+	timer.Stop()
			604	+	}
			605	+	}()
			606	+	return deadline
			607	+	}
			608	+
			609	+	func (d *snapshotShutdownDeadline) stop() {
			610	+	if d != nil {
			611	+	d.stopOnce.Do(func() {
			612	+	close(d.stopCh)
			613	+	<-d.finished
			614	+	})
582		}	615		}
583	-	d.snapshotWorkerMu.Lock()	616	+	}
584	-	cancel := d.snapshotWorkerCancel	617	+
585	-	if cancel == nil \|\| d.snapshotWorkerClosing {	618	+	func (d *snapshotShutdownDeadline) Done() <-chan struct{} {
586	-	d.snapshotWorkerMu.Unlock()	619	+	if d == nil {
587	-	return	620	+	return nil
588		}	621		}
589	-	d.snapshotWorkerClosing = true	622	+	return d.done
590	-	cancel()
591	-	d.snapshotWorkerMu.Unlock()
592	-	d.finishStoppedSnapshotWorker(true)
593		}	623		}
594			624
595		func (d Daemon) sessionsSnapshotLocked() []session {	625		func (d Daemon) sessionsSnapshotLocked() []session {

M internal/usecase/daemon/session.go +29 -6


@@ -879,6 +879,14 @@ func (d *Daemon) retryStoppedPurge(name string) error {
879	879		}
880	880
881	881		func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
	882	+	return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)
	883	+	}
	884	+
	885	+	// killSessionWithSnapshotDeadline shares Serve's shutdown budget with its
	886	+	// terminal checkpoint and coordinator join. A timed-out repository call keeps
	887	+	// only immutable capture state; it never observes closed worker channels or
	888	+	// session-owned cache that teardown has released.
	889	+	func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, purge bool, deadline *snapshotShutdownDeadline) error {
882	890		ringing := sess.anyAttention()
883	891		sess.mu.Lock()
884	892		isEphemeral := sess.ephemeral

@@ -912,7 +920,7 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
912	920		sess.snapshotMu.Lock()
913	921		terminalGeneration := sess.snapshotGeneration
914	922		sess.snapshotMu.Unlock()
915		-	if !d.scheduleFinalSnapshot(sess) \|\| !d.waitForSnapshotGeneration(sess, terminalGeneration) {
	923	+	if !d.scheduleFinalSnapshot(sess) \|\| !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
916	924		sess.mu.Lock()
917	925		name := sess.name
918	926		sess.mu.Unlock()

@@ -926,11 +934,26 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
926	934		}
927	935		if !isEphemeral {
928	936		// Joining first preserves byte ownership for a publication that is still
929		-	// encoding while this session is being torn down.
930		-	<-quarantineSnapshotCoordinator(sess)
931		-	sess.snapshotMu.Lock()
932		-	sess.snapshotChunkCache = nil
933		-	sess.snapshotMu.Unlock()
	937	+	// encoding while this session is being torn down. Once Serve's shared
	938	+	// budget expires, the worker is cancelled by its one final join instead;
	939	+	// keep the cache reachable because an uncooperative writer may still be
	940	+	// encoding immutable state.
	941	+	quarantineDone := quarantineSnapshotCoordinator(sess)
	942	+	joined := true
	943	+	if deadline != nil {
	944	+	select {
	945	+	case <-quarantineDone:
	946	+	case <-deadline.Done():
	947	+	joined = false
	948	+	}
	949	+	} else {
	950	+	<-quarantineDone
	951	+	}
	952	+	if joined {
	953	+	sess.snapshotMu.Lock()
	954	+	sess.snapshotChunkCache = nil
	955	+	sess.snapshotMu.Unlock()
	956	+	}
934	957		}
935	958		d.mu.Lock()
936	959		if _, ok := d.sessions[sess.id]; !ok {

 }
 func (d *Daemon) killSession(sess *session, reason uint8, purge bool) error {
+	return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)
+}
+// killSessionWithSnapshotDeadline shares Serve's shutdown budget with its
+// terminal checkpoint and coordinator join. A timed-out repository call keeps
+// only immutable capture state; it never observes closed worker channels or
+// session-owned cache that teardown has released.
+func (d *Daemon) killSessionWithSnapshotDeadline(sess *session, reason uint8, purge bool, deadline *snapshotShutdownDeadline) error {
 	ringing := sess.anyAttention()
 	sess.mu.Lock()
 	isEphemeral := sess.ephemeral
 			sess.snapshotMu.Lock()
 			terminalGeneration := sess.snapshotGeneration
 			sess.snapshotMu.Unlock()
-			if !d.scheduleFinalSnapshot(sess) || !d.waitForSnapshotGeneration(sess, terminalGeneration) {
+			if !d.scheduleFinalSnapshot(sess) || !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
 				sess.mu.Lock()
 				name := sess.name
 				sess.mu.Unlock()
 	}
 	if !isEphemeral {
 		// Joining first preserves byte ownership for a publication that is still
-		// encoding while this session is being torn down.
+		// encoding while this session is being torn down. Once Serve's shared
-		<-quarantineSnapshotCoordinator(sess)
+		// budget expires, the worker is cancelled by its one final join instead;
-		sess.snapshotMu.Lock()
+		// keep the cache reachable because an uncooperative writer may still be
-		sess.snapshotChunkCache = nil
+		// encoding immutable state.
-		sess.snapshotMu.Unlock()
+		quarantineDone := quarantineSnapshotCoordinator(sess)
+		joined := true
+		if deadline != nil {
+			select {
+			case <-quarantineDone:
+			case <-deadline.Done():
+				joined = false
+			}
+		} else {
+			<-quarantineDone
+		}
+		if joined {
+			sess.snapshotMu.Lock()
+			sess.snapshotChunkCache = nil
+			sess.snapshotMu.Unlock()
+		}
 	}
 	d.mu.Lock()
 	if _, ok := d.sessions[sess.id]; !ok {

M internal/usecase/daemon/snapshot.go +35 -7


@@ -568,8 +568,21 @@ func (d Daemon) waitForSnapshotGeneration(sess session, generation uint64) boo
568	568		if d == nil \|\| sess == nil \|\| generation == 0 {
569	569		return true
570	570		}
571		-	timer := d.clock.NewTimer(snapshotFinalFlushTimeout)
572		-	defer timer.Stop()
	571	+	deadline := newSnapshotShutdownDeadline(d.clock)
	572	+	defer deadline.stop()
	573	+	return d.waitForSnapshotGenerationWithDeadline(sess, generation, deadline)
	574	+	}
	575	+
	576	+	// waitForSnapshotGenerationWithDeadline waits for a forced checkpoint without
	577	+	// allocating another shutdown interval. Serve shares one deadline across every
	578	+	// session and the subsequent worker drain.
	579	+	func (d Daemon) waitForSnapshotGenerationWithDeadline(sess session, generation uint64, deadline *snapshotShutdownDeadline) bool {
	580	+	if d == nil \|\| sess == nil \|\| generation == 0 {
	581	+	return true
	582	+	}
	583	+	if deadline == nil {
	584	+	return d.waitForSnapshotGeneration(sess, generation)
	585	+	}
573	586		for {
574	587		sess.snapshotMu.Lock()
575	588		published := sess.snapshotPublishedGeneration >= generation

@@ -580,7 +593,7 @@ func (d Daemon) waitForSnapshotGeneration(sess session, generation uint64) boo
580	593		}
581	594		select {
582	595		case <-changed:
583		-	case <-timer.C():
	596	+	case <-deadline.Done():
584	597		return false
585	598		}
586	599		}

@@ -1012,6 +1025,19 @@ func (d Daemon) clearSnapshotWorkerInFlight(workerID uint64, capture snapshotC
1012	1025		}
1013	1026
1014	1027		func (d *Daemon) stopSnapshotEncodeWorker() {
	1028	+	if d == nil {
	1029	+	return
	1030	+	}
	1031	+	deadline := newSnapshotShutdownDeadline(d.clock)
	1032	+	defer deadline.stop()
	1033	+	d.stopSnapshotEncodeWorkerWithDeadline(deadline)
	1034	+	}
	1035	+
	1036	+	// stopSnapshotEncodeWorkerWithDeadline performs the one deadline-aware join
	1037	+	// for a Serve shutdown. Queues are never closed, and detached writers only
	1038	+	// retain immutable captures plus contexts/channels that remain live, so an
	1039	+	// uncooperative repository call cannot race teardown or block process exit.
	1040	+	func (d Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline snapshotShutdownDeadline) {
1015	1041		if d == nil {
1016	1042		return
1017	1043		}

@@ -1023,19 +1049,21 @@ func (d *Daemon) stopSnapshotEncodeWorker() {
1023	1049		}
1024	1050		// Stop accepting producers, then let the single worker drain its bounded
1025	1051		// queue. A non-context-aware store may still block indefinitely, so the
1026		-	// injected clock bounds how long teardown waits before detaching it.
	1052	+	// shared shutdown deadline bounds the one and only join.
1027	1053		d.snapshotWorkerClosing = true
1028	1054		flush := d.snapshotWorkerFlush
1029	1055		done := d.snapshotWorkerDone
1030	1056		close(flush)
1031		-	timer := d.clock.NewTimer(snapshotFinalFlushTimeout)
1032	1057		d.snapshotWorkerMu.Unlock()
1033	1058
	1059	+	if deadline == nil {
	1060	+	deadline = newSnapshotShutdownDeadline(d.clock)
	1061	+	defer deadline.stop()
	1062	+	}
1034	1063		select {
1035	1064		case <-done:
1036		-	timer.Stop()
1037	1065		d.finishStoppedSnapshotWorker(false)
1038		-	case <-timer.C():
	1066	+	case <-deadline.Done():
1039	1067		cancel()
1040	1068		d.finishStoppedSnapshotWorker(true)
1041	1069		}

 	if d == nil || sess == nil || generation == 0 {
 		return true
 	}
-	timer := d.clock.NewTimer(snapshotFinalFlushTimeout)
+	deadline := newSnapshotShutdownDeadline(d.clock)
-	defer timer.Stop()
+	defer deadline.stop()
+	return d.waitForSnapshotGenerationWithDeadline(sess, generation, deadline)
+}
+// waitForSnapshotGenerationWithDeadline waits for a forced checkpoint without
+// allocating another shutdown interval. Serve shares one deadline across every
+// session and the subsequent worker drain.
+func (d *Daemon) waitForSnapshotGenerationWithDeadline(sess *session, generation uint64, deadline *snapshotShutdownDeadline) bool {
+	if d == nil || sess == nil || generation == 0 {
+		return true
+	}
+	if deadline == nil {
+		return d.waitForSnapshotGeneration(sess, generation)
+	}
 	for {
 		sess.snapshotMu.Lock()
 		published := sess.snapshotPublishedGeneration >= generation
 		}
 		select {
 		case <-changed:
-		case <-timer.C():
+		case <-deadline.Done():
 			return false
 		}
 	}
 }
 func (d *Daemon) stopSnapshotEncodeWorker() {
+	if d == nil {
+		return
+	}
+	deadline := newSnapshotShutdownDeadline(d.clock)
+	defer deadline.stop()
+	d.stopSnapshotEncodeWorkerWithDeadline(deadline)
+}
+// stopSnapshotEncodeWorkerWithDeadline performs the one deadline-aware join
+// for a Serve shutdown. Queues are never closed, and detached writers only
+// retain immutable captures plus contexts/channels that remain live, so an
+// uncooperative repository call cannot race teardown or block process exit.
+func (d *Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline *snapshotShutdownDeadline) {
 	if d == nil {
 		return
 	}
 	}
 	// Stop accepting producers, then let the single worker drain its bounded
 	// queue. A non-context-aware store may still block indefinitely, so the
-	// injected clock bounds how long teardown waits before detaching it.
+	// shared shutdown deadline bounds the one and only join.
 	d.snapshotWorkerClosing = true
 	flush := d.snapshotWorkerFlush
 	done := d.snapshotWorkerDone
 	close(flush)
-	timer := d.clock.NewTimer(snapshotFinalFlushTimeout)
 	d.snapshotWorkerMu.Unlock()
+	if deadline == nil {
+		deadline = newSnapshotShutdownDeadline(d.clock)
+		defer deadline.stop()
+	}
 	select {
 	case <-done:
-		timer.Stop()
 		d.finishStoppedSnapshotWorker(false)
-	case <-timer.C():
+	case <-deadline.Done():
 		cancel()
 		d.finishStoppedSnapshotWorker(true)
 	}

M internal/usecase/daemon/snapshot_scheduler_test.go +15 -3


@@ -484,11 +484,13 @@ func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository
484	484		case <-time.After(testWaitTimeout):
485	485		t.Fatal("Serve waited beyond the terminal checkpoint deadline")
486	486		}
	487	+	require.LessOrEqual(t, clock.finalBudget.Load(), int64(snapshotFinalFlushTimeout), "shutdown must consume at most one snapshot deadline")
487	488		require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")
488	489		}
489	490
490	491		type serveShutdownClock struct {
491	492		finalTimers chan *snapshotDeadlineTimer
	493	+	finalBudget atomic.Int64
492	494		}
493	495
494	496		func newServeShutdownClock() *serveShutdownClock {

@@ -500,7 +502,9 @@ func (c *serveShutdownClock) NewTimer(delay time.Duration) ports.Timer {
500	502		if delay != snapshotFinalFlushTimeout {
501	503		return stubTimer{}
502	504		}
503		-	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
	505	+	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1), onFire: func() {
	506	+	c.finalBudget.Add(int64(delay))
	507	+	}}
504	508		c.finalTimers <- timer
505	509		return timer
506	510		}

@@ -612,12 +616,20 @@ func (c snapshotDeadlineClock) nextTimer(t testing.T) *snapshotDeadlineTimer {
612	616		}
613	617		}
614	618
615		-	type snapshotDeadlineTimer struct{ ch chan time.Time }
	619	+	type snapshotDeadlineTimer struct {
	620	+	ch chan time.Time
	621	+	onFire func()
	622	+	}
616	623
617	624		func (t *snapshotDeadlineTimer) C() <-chan time.Time { return t.ch }
618	625		func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true }
619	626		func (*snapshotDeadlineTimer) Stop() bool { return true }
620		-	func (t *snapshotDeadlineTimer) fire() { t.ch <- time.Unix(101, 0) }
	627	+	func (t *snapshotDeadlineTimer) fire() {
	628	+	if t.onFire != nil {
	629	+	t.onFire()
	630	+	}
	631	+	t.ch <- time.Unix(101, 0)
	632	+	}
621	633
622	634		// deterministicSnapshotClock records timer resets and advances only when a
623	635		// test explicitly fires its timer. It deliberately has no wall-clock behavior.

 	case <-time.After(testWaitTimeout):
 		t.Fatal("Serve waited beyond the terminal checkpoint deadline")
 	}
+	require.LessOrEqual(t, clock.finalBudget.Load(), int64(snapshotFinalFlushTimeout), "shutdown must consume at most one snapshot deadline")
 	require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")
 }
 type serveShutdownClock struct {
 	finalTimers chan *snapshotDeadlineTimer
+	finalBudget atomic.Int64
 }
 func newServeShutdownClock() *serveShutdownClock {
 	if delay != snapshotFinalFlushTimeout {
 		return stubTimer{}
 	}
-	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
+	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1), onFire: func() {
+		c.finalBudget.Add(int64(delay))
+	}}
 	c.finalTimers <- timer
 	return timer
 }
 	}
 }
-type snapshotDeadlineTimer struct{ ch chan time.Time }
+type snapshotDeadlineTimer struct {
+	ch     chan time.Time
+	onFire func()
+}
 func (t *snapshotDeadlineTimer) C() <-chan time.Time    { return t.ch }
 func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true }
 func (*snapshotDeadlineTimer) Stop() bool               { return true }
-func (t *snapshotDeadlineTimer) fire()                  { t.ch <- time.Unix(101, 0) }
+func (t *snapshotDeadlineTimer) fire() {
+	if t.onFire != nil {
+		t.onFire()
+	}
+	t.ch <- time.Unix(101, 0)
+}
 // deterministicSnapshotClock records timer resets and advances only when a
 // test explicitly fires its timer. It deliberately has no wall-clock behavior.