fix(daemon): coordinate incremental checkpoints

commit: d32ee68665fd779dfa3445cb85ff3044417d266d
author: brice · 3mo ago · 2026-07-21 22:18 UTC
committer: brice · 3mo ago · 2026-07-21 22:19 UTC
parent: 73a5ab3
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +262 -32

M internal/usecase/daemon/session.go +65 -16


@@ -43,6 +43,7 @@ type session struct {
43	43		client *attachedClient
44	44		clipboardQueue []clipboardForward
45	45		clipboardWorkerRunning bool
	46	+	renameInProgress bool
46	47		cwd string
47	48		terminal terminalEnv
48	49		// env is the authoritative immutable environment snapshot for future PTY children.

@@ -61,8 +62,18 @@ type session struct {
61	62		snapshotNextEligibleAt time.Time
62	63		snapshotAttempted bool
63	64		snapshotAttemptKind snapshotAttemptKind
	65	+	// The coordinator state below is guarded by snapshotMu. A capture can be
	66	+	// queued globally or in flight, never more than one of each per session.
	67	+	// Quarantine cancels the session publication context before destructive
	68	+	// repository operations and publicationDone joins a started publication.
64	69		snapshotPending bool
65	70		snapshotPendingCaptures uint
	71	+	snapshotQueuedCapture *snapshotCapture
	72	+	snapshotInFlightCapture *snapshotCapture
	73	+	snapshotPublicationDone chan struct{}
	74	+	snapshotPublicationContext context.Context
	75	+	snapshotPublicationCancel context.CancelFunc
	76	+	snapshotQuarantined bool
66	77		// snapshotFailureSig is the stable class of the active failed publication.
67	78		// It is guarded by snapshotMu and deliberately never contains error text.
68	79		snapshotFailureSig string

@@ -530,15 +541,27 @@ func (d Daemon) renameSession(sess session, name string) error {
530	541		if err := domain.ValidateSessionName(name); err != nil {
531	542		return err
532	543		}
	544	+
	545	+	// Reserve the rename while state is inspected, then perform persistence and
	546	+	// repository I/O without daemon/session locks. In particular, Delete must
	547	+	// follow coordinator quarantine+join or a late Publish could recreate the
	548	+	// old repository identity.
533	549		d.mu.Lock()
534		-	defer d.mu.Unlock()
535	550		if taken := d.findByNameLocked(name); taken != nil && taken != sess {
	551	+	d.mu.Unlock()
536	552		return errSessionNameInUse
537	553		}
	554	+	sess.mu.Lock()
	555	+	if sess.renameInProgress {
	556	+	sess.mu.Unlock()
	557	+	d.mu.Unlock()
	558	+	return errors.New("session rename already in progress")
	559	+	}
538	560		if stopped, ok := d.stopped[name]; ok && stopped.name != sess.name {
	561	+	sess.mu.Unlock()
	562	+	d.mu.Unlock()
539	563		return errSessionNameInUse
540	564		}
541		-	sess.mu.Lock()
542	565		oldName := sess.name
543	566		wasEphemeral := sess.ephemeral
544	567		createdAt := sess.createdAt

@@ -547,45 +570,65 @@ func (d Daemon) renameSession(sess session, name string) error {
547	570		createdAt, err = d.allocateLifecycleCreatedAtLocked()
548	571		if err != nil {
549	572		sess.mu.Unlock()
	573	+	d.mu.Unlock()
550	574		return err
551	575		}
552	576		}
553	577		lastUsedSeq := sess.mruAt.Load()
	578	+	record := sess.persistRecordLocked(d.nowUnixNano())
	579	+	record.Name = name
	580	+	record.CreatedAt = createdAt
	581	+	record.LastUsedSeq = lastUsedSeq
	582	+	sess.renameInProgress = true
	583	+	sess.mu.Unlock()
	584	+	d.mu.Unlock()
	585	+
	586	+	rollback := func(err error, resume bool) error {
	587	+	if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
	588	+	d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
	589	+	}
	590	+	if resume {
	591	+	resumeSnapshotCoordinator(sess)
	592	+	}
	593	+	sess.mu.Lock()
	594	+	sess.renameInProgress = false
	595	+	sess.mu.Unlock()
	596	+	return err
	597	+	}
554	598		if wasEphemeral \|\| oldName != name {
555		-	record := sess.persistRecordLocked(d.nowUnixNano())
556		-	record.Name = name
557		-	record.CreatedAt = createdAt
558		-	record.LastUsedSeq = lastUsedSeq
559	599		if err := d.persist.Save(record); err != nil {
	600	+	sess.mu.Lock()
	601	+	sess.renameInProgress = false
560	602		sess.mu.Unlock()
561	603		return err
562	604		}
563	605		}
564	606		if !wasEphemeral && oldName != name {
	607	+	<-quarantineSnapshotCoordinator(sess)
565	608		if d.snapsEnabled && d.snapshotRepository != nil {
566	609		if err := d.snapshotRepository.Delete(context.Background(), oldName); err != nil {
567		-	if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
568		-	d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
569		-	}
570		-	sess.mu.Unlock()
571		-	return err
	610	+	return rollback(err, true)
572	611		}
573	612		}
574	613		if err := d.persist.Delete(oldName); err != nil {
575		-	if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
576		-	d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
577		-	}
578		-	sess.mu.Unlock()
579		-	return err
	614	+	return rollback(err, true)
580	615		}
	616	+	resumeSnapshotCoordinator(sess)
581	617		}
	618	+
	619	+	d.mu.Lock()
	620	+	sess.mu.Lock()
	621	+	// The reservation prevents another rename. A closing session is allowed to
	622	+	// finish this short commit; killSession subsequently owns its teardown.
582	623		delete(d.stopped, oldName)
583	624		delete(d.stopped, name)
584	625		sess.name = name
585	626		sess.createdAt = createdAt
586	627		sess.ephemeral = false
	628	+	sess.renameInProgress = false
587	629		sess.snapEligible.Store(name != "")
588	630		sess.mu.Unlock()
	631	+	d.mu.Unlock()
589	632		markSnapshotDirty(sess)
590	633		return nil
591	634		}

@@ -731,6 +774,12 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
731	774		sess.mu.Lock()
732	775		isEphemeral := sess.ephemeral
733	776		sess.mu.Unlock()
	777	+	// A repository delete must never race an older publication that could
	778	+	// recreate the just-quarantined name. Do not hold daemon or session locks
	779	+	// while joining: a repository implementation may ignore cancellation.
	780	+	if purge && !isEphemeral && d.snapshotRepository != nil {
	781	+	<-quarantineSnapshotCoordinator(sess)
	782	+	}
734	783		if !purge && !isEphemeral && d.persistEnabled {
735	784		d.refreshSessionCwd(sess)
736	785		}

 	client                 *attachedClient
 	clipboardQueue         []clipboardForward
 	clipboardWorkerRunning bool
+	renameInProgress       bool
 	cwd                    string
 	terminal               terminalEnv
 	// env is the authoritative immutable environment snapshot for future PTY children.
 	snapshotNextEligibleAt     time.Time
 	snapshotAttempted          bool
 	snapshotAttemptKind        snapshotAttemptKind
+	// The coordinator state below is guarded by snapshotMu. A capture can be
+	// queued globally or in flight, never more than one of each per session.
+	// Quarantine cancels the session publication context before destructive
+	// repository operations and publicationDone joins a started publication.
 	snapshotPending            bool
 	snapshotPendingCaptures    uint
+	snapshotQueuedCapture      *snapshotCapture
+	snapshotInFlightCapture    *snapshotCapture
+	snapshotPublicationDone    chan struct{}
+	snapshotPublicationContext context.Context
+	snapshotPublicationCancel  context.CancelFunc
+	snapshotQuarantined        bool
 	// snapshotFailureSig is the stable class of the active failed publication.
 	// It is guarded by snapshotMu and deliberately never contains error text.
 	snapshotFailureSig string
 	if err := domain.ValidateSessionName(name); err != nil {
 		return err
 	}
+	// Reserve the rename while state is inspected, then perform persistence and
+	// repository I/O without daemon/session locks. In particular, Delete must
+	// follow coordinator quarantine+join or a late Publish could recreate the
+	// old repository identity.
 	d.mu.Lock()
-	defer d.mu.Unlock()
 	if taken := d.findByNameLocked(name); taken != nil && taken != sess {
+		d.mu.Unlock()
 		return errSessionNameInUse
 	}
+	sess.mu.Lock()
+	if sess.renameInProgress {
+		sess.mu.Unlock()
+		d.mu.Unlock()
+		return errors.New("session rename already in progress")
+	}
 	if stopped, ok := d.stopped[name]; ok && stopped.name != sess.name {
+		sess.mu.Unlock()
+		d.mu.Unlock()
 		return errSessionNameInUse
 	}
-	sess.mu.Lock()
 	oldName := sess.name
 	wasEphemeral := sess.ephemeral
 	createdAt := sess.createdAt
 		createdAt, err = d.allocateLifecycleCreatedAtLocked()
 		if err != nil {
 			sess.mu.Unlock()
+			d.mu.Unlock()
 			return err
 		}
 	}
 	lastUsedSeq := sess.mruAt.Load()
+	record := sess.persistRecordLocked(d.nowUnixNano())
+	record.Name = name
+	record.CreatedAt = createdAt
+	record.LastUsedSeq = lastUsedSeq
+	sess.renameInProgress = true
+	sess.mu.Unlock()
+	d.mu.Unlock()
+	rollback := func(err error, resume bool) error {
+		if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
+			d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
+		}
+		if resume {
+			resumeSnapshotCoordinator(sess)
+		}
+		sess.mu.Lock()
+		sess.renameInProgress = false
+		sess.mu.Unlock()
+		return err
+	}
 	if wasEphemeral || oldName != name {
-		record := sess.persistRecordLocked(d.nowUnixNano())
-		record.Name = name
-		record.CreatedAt = createdAt
-		record.LastUsedSeq = lastUsedSeq
 		if err := d.persist.Save(record); err != nil {
+			sess.mu.Lock()
+			sess.renameInProgress = false
 			sess.mu.Unlock()
 			return err
 		}
 	}
 	if !wasEphemeral && oldName != name {
+		<-quarantineSnapshotCoordinator(sess)
 		if d.snapsEnabled && d.snapshotRepository != nil {
 			if err := d.snapshotRepository.Delete(context.Background(), oldName); err != nil {
-				if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
+				return rollback(err, true)
-					d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
-				}
-				sess.mu.Unlock()
-				return err
 			}
 		}
 		if err := d.persist.Delete(oldName); err != nil {
-			if cleanupErr := d.persist.Delete(name); cleanupErr != nil {
+			return rollback(err, true)
-				d.log.Warn("cleaning up renamed persisted session failed", "err", cleanupErr, "session", name)
-			}
-			sess.mu.Unlock()
-			return err
 		}
+		resumeSnapshotCoordinator(sess)
 	}
+	d.mu.Lock()
+	sess.mu.Lock()
+	// The reservation prevents another rename. A closing session is allowed to
+	// finish this short commit; killSession subsequently owns its teardown.
 	delete(d.stopped, oldName)
 	delete(d.stopped, name)
 	sess.name = name
 	sess.createdAt = createdAt
 	sess.ephemeral = false
+	sess.renameInProgress = false
 	sess.snapEligible.Store(name != "")
 	sess.mu.Unlock()
+	d.mu.Unlock()
 	markSnapshotDirty(sess)
 	return nil
 }
 	sess.mu.Lock()
 	isEphemeral := sess.ephemeral
 	sess.mu.Unlock()
+	// A repository delete must never race an older publication that could
+	// recreate the just-quarantined name. Do not hold daemon or session locks
+	// while joining: a repository implementation may ignore cancellation.
+	if purge && !isEphemeral && d.snapshotRepository != nil {
+		<-quarantineSnapshotCoordinator(sess)
+	}
 	if !purge && !isEphemeral && d.persistEnabled {
 		d.refreshSessionCwd(sess)
 	}

M internal/usecase/daemon/snapshot.go +143 -16


@@ -34,14 +34,16 @@ const (
34	34		)
35	35
36	36		type snapshotCapture struct {
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
	37	+	session *session
	38	+	attemptKind snapshotAttemptKind
	39	+	generation uint64
	40	+	name string
	41	+	createdAt uint64
	42	+	active uint16
	43	+	tabs []snapshotCaptureTab
	44	+	publicationContext context.Context
	45	+	coordinatorDiscarded bool // guarded by session.snapshotMu
	46	+	finishOnce sync.Once
45	47		}
46	48
47	49		type snapshotCaptureTab struct {

@@ -299,6 +301,72 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err
299	301		return nil
300	302		}
301	303
	304	+	// snapshotCoordinatorContext returns the cancellation context shared by this
	305	+	// session's queued and in-flight captures. snapshotMu serializes creation with
	306	+	// quarantine, so a publisher can never obtain a live context after quarantine.
	307	+	func snapshotCoordinatorContext(sess *session) (context.Context, context.CancelFunc) {
	308	+	sess.snapshotMu.Lock()
	309	+	defer sess.snapshotMu.Unlock()
	310	+	if sess.snapshotPublicationContext == nil {
	311	+	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
	312	+	}
	313	+	return sess.snapshotPublicationContext, sess.snapshotPublicationCancel
	314	+	}
	315	+
	316	+	// quarantineSnapshotCoordinator prevents a deleted session from starting any
	317	+	// further publication. The returned join channel is closed after an already
	318	+	// started repository call returns; callers must wait without daemon or session
	319	+	// locks before deleting repository state.
	320	+	func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
	321	+	if sess == nil {
	322	+	done := make(chan struct{})
	323	+	close(done)
	324	+	return done
	325	+	}
	326	+	sess.snapshotMu.Lock()
	327	+	sess.snapshotQuarantined = true
	328	+	sess.snapEligible.Store(false)
	329	+	if sess.snapshotPublicationCancel != nil {
	330	+	sess.snapshotPublicationCancel()
	331	+	}
	332	+	// Remove a not-yet-started capture from the coordinator immediately. The
	333	+	// global queue may still contain it, but startSnapshotPublication compares
	334	+	// identity and will discard it even if this session is later renamed.
	335	+	if sess.snapshotQueuedCapture != nil {
	336	+	sess.snapshotQueuedCapture.coordinatorDiscarded = true
	337	+	sess.snapshotQueuedCapture = nil
	338	+	if sess.snapshotPendingCaptures > 0 {
	339	+	sess.snapshotPendingCaptures--
	340	+	}
	341	+	sess.snapshotPending = sess.snapshotPendingCaptures > 0 \|\| sess.snapshotInFlightCapture != nil
	342	+	}
	343	+	done := sess.snapshotPublicationDone
	344	+	if done == nil {
	345	+	done = make(chan struct{})
	346	+	close(done)
	347	+	}
	348	+	sess.signalSnapshotChangedLocked()
	349	+	sess.snapshotMu.Unlock()
	350	+	return done
	351	+	}
	352	+
	353	+	// resumeSnapshotCoordinator is used only after a rename's old repository name
	354	+	// has been quarantined and deleted. Queued captures were discarded during
	355	+	// quarantine, so the new context cannot publish under the old identity.
	356	+	func resumeSnapshotCoordinator(sess *session) {
	357	+	if sess == nil {
	358	+	return
	359	+	}
	360	+	sess.snapshotMu.Lock()
	361	+	if sess.snapshotPublicationCancel != nil {
	362	+	sess.snapshotPublicationCancel()
	363	+	}
	364	+	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
	365	+	sess.snapshotQuarantined = false
	366	+	sess.signalSnapshotChangedLocked()
	367	+	sess.snapshotMu.Unlock()
	368	+	}
	369	+
302	370		func markSnapshotDirty(sess *session) {
303	371		if sess == nil \|\| !sess.snapEligible.Load() {
304	372		return

@@ -395,11 +463,20 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
395	463		}
396	464		sess.snapshotMu.Lock()
397	465		routineRateLimited := d.snapshotRepository != nil
398		-	if !sess.snapDirty.Load() \|\| (!final && (sess.snapshotPending \|\| (routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)))) {
	466	+	// Forced attempts coalesce with a routine capture too. This keeps the
	467	+	// coordinator at one queued plus one in-flight capture, while the dirty
	468	+	// generation ensures a later checkpoint observes mutations made meanwhile.
	469	+	if sess.snapshotQuarantined \|\| !sess.snapDirty.Load() \|\| sess.snapshotPending \|\|
	470	+	(!final && routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
	471	+	available := !sess.snapshotQuarantined
399	472		sess.snapshotMu.Unlock()
400		-	return true
	473	+	return available
	474	+	}
	475	+	if sess.snapshotPublicationContext == nil {
	476	+	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
401	477		}
402	478		generation := sess.snapshotGeneration
	479	+	publicationContext := sess.snapshotPublicationContext
403	480		sess.snapshotCapturedGeneration = generation
404	481		sess.snapshotAttempted = true
405	482		sess.snapshotAttemptKind = kind

@@ -411,11 +488,22 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
411	488		capture, ok := d.captureSnapshotState(sess, generation)
412	489		if capture != nil {
413	490		capture.attemptKind = kind
	491	+	capture.publicationContext = publicationContext
414	492		}
415	493		if !ok {
416	494		d.finishSnapshotCapture(capture, false)
417	495		return false
418	496		}
	497	+	sess.snapshotMu.Lock()
	498	+	quarantined := sess.snapshotQuarantined
	499	+	if !quarantined {
	500	+	sess.snapshotQueuedCapture = capture
	501	+	}
	502	+	sess.snapshotMu.Unlock()
	503	+	if quarantined {
	504	+	d.finishSnapshotCapture(capture, false)
	505	+	return false
	506	+	}
419	507		if d.enqueueSnapshotCapture(capture) \|\| (final && d.enqueueFinalSnapshotCapture(capture)) {
420	508		return true
421	509		}

@@ -586,6 +674,28 @@ func (d Daemon) encodeSnapshotCapture(capture snapshotCapture) ([]byte, error)
586	674		return d.snapshotMarshal(snap)
587	675		}
588	676
	677	+	// startSnapshotPublication moves a retained capture from the producer queue to
	678	+	// the repository worker. Quarantine and this transition share snapshotMu, so a
	679	+	// repository call cannot begin after destructive teardown has quarantined the
	680	+	// session.
	681	+	func startSnapshotPublication(capture *snapshotCapture) bool {
	682	+	if capture == nil \|\| capture.session == nil {
	683	+	return false
	684	+	}
	685	+	sess := capture.session
	686	+	sess.snapshotMu.Lock()
	687	+	defer sess.snapshotMu.Unlock()
	688	+	if sess.snapshotQuarantined \|\| sess.snapshotQueuedCapture != capture {
	689	+	return false
	690	+	}
	691	+	sess.snapshotQueuedCapture = nil
	692	+	sess.snapshotInFlightCapture = capture
	693	+	sess.snapshotPublicationDone = make(chan struct{})
	694	+	sess.snapshotPending = true
	695	+	sess.signalSnapshotChangedLocked()
	696	+	return true
	697	+	}
	698	+
589	699		func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool) {
590	700		if capture == nil {
591	701		return

@@ -595,8 +705,19 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
595	705		if capture.session.snapshotPendingCaptures > 0 {
596	706		capture.session.snapshotPendingCaptures--
597	707		}
598		-	capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0
599		-	if capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {
	708	+	if capture.session.snapshotQueuedCapture == capture {
	709	+	capture.session.snapshotQueuedCapture = nil
	710	+	}
	711	+	if capture.session.snapshotInFlightCapture == capture {
	712	+	capture.session.snapshotInFlightCapture = nil
	713	+	if capture.session.snapshotPublicationDone != nil {
	714	+	close(capture.session.snapshotPublicationDone)
	715	+	capture.session.snapshotPublicationDone = nil
	716	+	}
	717	+	}
	718	+	capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 \|\|
	719	+	capture.session.snapshotQueuedCapture != nil \|\| capture.session.snapshotInFlightCapture != nil
	720	+	if !capture.coordinatorDiscarded && capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {
600	721		// The interval is measured from completion, including a failed
601	722		// attempt. A forced teardown checkpoint deliberately leaves it alone.
602	723		capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)

@@ -695,16 +816,22 @@ func (d *Daemon) startSnapshotEncodeWorker() {
695	816		go func() {
696	817		defer close(done)
697	818		write := func(capture *snapshotCapture) bool {
698		-	if workerCtx.Err() != nil \|\| !d.setSnapshotWorkerInFlight(workerID, capture) {
	819	+	if workerCtx.Err() != nil \|\| !startSnapshotPublication(capture) \|\| !d.setSnapshotWorkerInFlight(workerID, capture) {
699	820		d.finishSnapshotCapture(capture, false)
700		-	return false
	821	+	return workerCtx.Err() == nil
701	822		}
702	823		publication, err := d.incrementalPublication(capture)
	824	+	publicationContext := capture.publicationContext
	825	+	if publicationContext == nil {
	826	+	publicationContext = workerCtx
	827	+	}
703	828		if err == nil && workerCtx.Err() == nil {
704		-	err = d.snapshotRepository.Publish(workerCtx, publication)
	829	+	if err = publicationContext.Err(); err == nil {
	830	+	err = d.snapshotRepository.Publish(publicationContext, publication)
	831	+	}
705	832		}
706	833		d.clearSnapshotWorkerInFlight(workerID, capture)
707		-	if err != nil && workerCtx.Err() == nil {
	834	+	if err != nil && workerCtx.Err() == nil && publicationContext.Err() == nil {
708	835		// Global, not session-scoped: by now the session may already be
709	836		// torn down, so a session notice would be dead-on-arrival. No lock
710	837		// is held here (clearSnapshotWorkerInFlight released its own).


@@ -34,14 +34,16 @@ const (
34		)	34		)
35			35
36		type snapshotCapture struct {	36		type snapshotCapture struct {
37	-	session *session	37	+	session *session
38	-	attemptKind snapshotAttemptKind	38	+	attemptKind snapshotAttemptKind
39	-	generation uint64	39	+	generation uint64
40	-	name string	40	+	name string
41	-	createdAt uint64	41	+	createdAt uint64
42	-	active uint16	42	+	active uint16
43	-	tabs []snapshotCaptureTab	43	+	tabs []snapshotCaptureTab
44	-	finishOnce sync.Once	44	+	publicationContext context.Context
			45	+	coordinatorDiscarded bool // guarded by session.snapshotMu
			46	+	finishOnce sync.Once
45		}	47		}
46			48
47		type snapshotCaptureTab struct {	49		type snapshotCaptureTab struct {

@@ -299,6 +301,72 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err
299		return nil	301		return nil
300		}	302		}
301			303
			304	+	// snapshotCoordinatorContext returns the cancellation context shared by this
			305	+	// session's queued and in-flight captures. snapshotMu serializes creation with
			306	+	// quarantine, so a publisher can never obtain a live context after quarantine.
			307	+	func snapshotCoordinatorContext(sess *session) (context.Context, context.CancelFunc) {
			308	+	sess.snapshotMu.Lock()
			309	+	defer sess.snapshotMu.Unlock()
			310	+	if sess.snapshotPublicationContext == nil {
			311	+	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
			312	+	}
			313	+	return sess.snapshotPublicationContext, sess.snapshotPublicationCancel
			314	+	}
			315	+
			316	+	// quarantineSnapshotCoordinator prevents a deleted session from starting any
			317	+	// further publication. The returned join channel is closed after an already
			318	+	// started repository call returns; callers must wait without daemon or session
			319	+	// locks before deleting repository state.
			320	+	func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
			321	+	if sess == nil {
			322	+	done := make(chan struct{})
			323	+	close(done)
			324	+	return done
			325	+	}
			326	+	sess.snapshotMu.Lock()
			327	+	sess.snapshotQuarantined = true
			328	+	sess.snapEligible.Store(false)
			329	+	if sess.snapshotPublicationCancel != nil {
			330	+	sess.snapshotPublicationCancel()
			331	+	}
			332	+	// Remove a not-yet-started capture from the coordinator immediately. The
			333	+	// global queue may still contain it, but startSnapshotPublication compares
			334	+	// identity and will discard it even if this session is later renamed.
			335	+	if sess.snapshotQueuedCapture != nil {
			336	+	sess.snapshotQueuedCapture.coordinatorDiscarded = true
			337	+	sess.snapshotQueuedCapture = nil
			338	+	if sess.snapshotPendingCaptures > 0 {
			339	+	sess.snapshotPendingCaptures--
			340	+	}
			341	+	sess.snapshotPending = sess.snapshotPendingCaptures > 0 \|\| sess.snapshotInFlightCapture != nil
			342	+	}
			343	+	done := sess.snapshotPublicationDone
			344	+	if done == nil {
			345	+	done = make(chan struct{})
			346	+	close(done)
			347	+	}
			348	+	sess.signalSnapshotChangedLocked()
			349	+	sess.snapshotMu.Unlock()
			350	+	return done
			351	+	}
			352	+
			353	+	// resumeSnapshotCoordinator is used only after a rename's old repository name
			354	+	// has been quarantined and deleted. Queued captures were discarded during
			355	+	// quarantine, so the new context cannot publish under the old identity.
			356	+	func resumeSnapshotCoordinator(sess *session) {
			357	+	if sess == nil {
			358	+	return
			359	+	}
			360	+	sess.snapshotMu.Lock()
			361	+	if sess.snapshotPublicationCancel != nil {
			362	+	sess.snapshotPublicationCancel()
			363	+	}
			364	+	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
			365	+	sess.snapshotQuarantined = false
			366	+	sess.signalSnapshotChangedLocked()
			367	+	sess.snapshotMu.Unlock()
			368	+	}
			369	+
302		func markSnapshotDirty(sess *session) {	370		func markSnapshotDirty(sess *session) {
303		if sess == nil \|\| !sess.snapEligible.Load() {	371		if sess == nil \|\| !sess.snapEligible.Load() {
304		return	372		return

@@ -395,11 +463,20 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
395		}	463		}
396		sess.snapshotMu.Lock()	464		sess.snapshotMu.Lock()
397		routineRateLimited := d.snapshotRepository != nil	465		routineRateLimited := d.snapshotRepository != nil
398	-	if !sess.snapDirty.Load() \|\| (!final && (sess.snapshotPending \|\| (routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)))) {	466	+	// Forced attempts coalesce with a routine capture too. This keeps the
			467	+	// coordinator at one queued plus one in-flight capture, while the dirty
			468	+	// generation ensures a later checkpoint observes mutations made meanwhile.
			469	+	if sess.snapshotQuarantined \|\| !sess.snapDirty.Load() \|\| sess.snapshotPending \|\|
			470	+	(!final && routineRateLimited && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
			471	+	available := !sess.snapshotQuarantined
399		sess.snapshotMu.Unlock()	472		sess.snapshotMu.Unlock()
400	-	return true	473	+	return available
			474	+	}
			475	+	if sess.snapshotPublicationContext == nil {
			476	+	sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
401		}	477		}
402		generation := sess.snapshotGeneration	478		generation := sess.snapshotGeneration
			479	+	publicationContext := sess.snapshotPublicationContext
403		sess.snapshotCapturedGeneration = generation	480		sess.snapshotCapturedGeneration = generation
404		sess.snapshotAttempted = true	481		sess.snapshotAttempted = true
405		sess.snapshotAttemptKind = kind	482		sess.snapshotAttemptKind = kind

@@ -411,11 +488,22 @@ func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bo
411		capture, ok := d.captureSnapshotState(sess, generation)	488		capture, ok := d.captureSnapshotState(sess, generation)
412		if capture != nil {	489		if capture != nil {
413		capture.attemptKind = kind	490		capture.attemptKind = kind
			491	+	capture.publicationContext = publicationContext
414		}	492		}
415		if !ok {	493		if !ok {
416		d.finishSnapshotCapture(capture, false)	494		d.finishSnapshotCapture(capture, false)
417		return false	495		return false
418		}	496		}
			497	+	sess.snapshotMu.Lock()
			498	+	quarantined := sess.snapshotQuarantined
			499	+	if !quarantined {
			500	+	sess.snapshotQueuedCapture = capture
			501	+	}
			502	+	sess.snapshotMu.Unlock()
			503	+	if quarantined {
			504	+	d.finishSnapshotCapture(capture, false)
			505	+	return false
			506	+	}
419		if d.enqueueSnapshotCapture(capture) \|\| (final && d.enqueueFinalSnapshotCapture(capture)) {	507		if d.enqueueSnapshotCapture(capture) \|\| (final && d.enqueueFinalSnapshotCapture(capture)) {
420		return true	508		return true
421		}	509		}

@@ -586,6 +674,28 @@ func (d Daemon) encodeSnapshotCapture(capture snapshotCapture) ([]byte, error)
586		return d.snapshotMarshal(snap)	674		return d.snapshotMarshal(snap)
587		}	675		}
588			676
			677	+	// startSnapshotPublication moves a retained capture from the producer queue to
			678	+	// the repository worker. Quarantine and this transition share snapshotMu, so a
			679	+	// repository call cannot begin after destructive teardown has quarantined the
			680	+	// session.
			681	+	func startSnapshotPublication(capture *snapshotCapture) bool {
			682	+	if capture == nil \|\| capture.session == nil {
			683	+	return false
			684	+	}
			685	+	sess := capture.session
			686	+	sess.snapshotMu.Lock()
			687	+	defer sess.snapshotMu.Unlock()
			688	+	if sess.snapshotQuarantined \|\| sess.snapshotQueuedCapture != capture {
			689	+	return false
			690	+	}
			691	+	sess.snapshotQueuedCapture = nil
			692	+	sess.snapshotInFlightCapture = capture
			693	+	sess.snapshotPublicationDone = make(chan struct{})
			694	+	sess.snapshotPending = true
			695	+	sess.signalSnapshotChangedLocked()
			696	+	return true
			697	+	}
			698	+
589		func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool) {	699		func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool) {
590		if capture == nil {	700		if capture == nil {
591		return	701		return

@@ -595,8 +705,19 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
595		if capture.session.snapshotPendingCaptures > 0 {	705		if capture.session.snapshotPendingCaptures > 0 {
596		capture.session.snapshotPendingCaptures--	706		capture.session.snapshotPendingCaptures--
597		}	707		}
598	-	capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0	708	+	if capture.session.snapshotQueuedCapture == capture {
599	-	if capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {	709	+	capture.session.snapshotQueuedCapture = nil
			710	+	}
			711	+	if capture.session.snapshotInFlightCapture == capture {
			712	+	capture.session.snapshotInFlightCapture = nil
			713	+	if capture.session.snapshotPublicationDone != nil {
			714	+	close(capture.session.snapshotPublicationDone)
			715	+	capture.session.snapshotPublicationDone = nil
			716	+	}
			717	+	}
			718	+	capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 \|\|
			719	+	capture.session.snapshotQueuedCapture != nil \|\| capture.session.snapshotInFlightCapture != nil
			720	+	if !capture.coordinatorDiscarded && capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {
600		// The interval is measured from completion, including a failed	721		// The interval is measured from completion, including a failed
601		// attempt. A forced teardown checkpoint deliberately leaves it alone.	722		// attempt. A forced teardown checkpoint deliberately leaves it alone.
602		capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)	723		capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)

@@ -695,16 +816,22 @@ func (d *Daemon) startSnapshotEncodeWorker() {
695		go func() {	816		go func() {
696		defer close(done)	817		defer close(done)
697		write := func(capture *snapshotCapture) bool {	818		write := func(capture *snapshotCapture) bool {
698	-	if workerCtx.Err() != nil \|\| !d.setSnapshotWorkerInFlight(workerID, capture) {	819	+	if workerCtx.Err() != nil \|\| !startSnapshotPublication(capture) \|\| !d.setSnapshotWorkerInFlight(workerID, capture) {
699		d.finishSnapshotCapture(capture, false)	820		d.finishSnapshotCapture(capture, false)
700	-	return false	821	+	return workerCtx.Err() == nil
701		}	822		}
702		publication, err := d.incrementalPublication(capture)	823		publication, err := d.incrementalPublication(capture)
			824	+	publicationContext := capture.publicationContext
			825	+	if publicationContext == nil {
			826	+	publicationContext = workerCtx
			827	+	}
703		if err == nil && workerCtx.Err() == nil {	828		if err == nil && workerCtx.Err() == nil {
704	-	err = d.snapshotRepository.Publish(workerCtx, publication)	829	+	if err = publicationContext.Err(); err == nil {
			830	+	err = d.snapshotRepository.Publish(publicationContext, publication)
			831	+	}
705		}	832		}
706		d.clearSnapshotWorkerInFlight(workerID, capture)	833		d.clearSnapshotWorkerInFlight(workerID, capture)
707	-	if err != nil && workerCtx.Err() == nil {	834	+	if err != nil && workerCtx.Err() == nil && publicationContext.Err() == nil {
708		// Global, not session-scoped: by now the session may already be	835		// Global, not session-scoped: by now the session may already be
709		// torn down, so a session notice would be dead-on-arrival. No lock	836		// torn down, so a session notice would be dead-on-arrival. No lock
710		// is held here (clearSnapshotWorkerInFlight released its own).	837		// is held here (clearSnapshotWorkerInFlight released its own).

M internal/usecase/daemon/snapshot_scheduler_test.go +54 -0


@@ -14,6 +14,60 @@ import (
14	14		portsmocks "github.com/bnema/vev/internal/ports/mocks"
15	15		)
16	16
	17	+	func TestSnapshotCoordinatorQuarantineCancelsPublicationBeforeDelete(t *testing.T) {
	18	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	19	+	ctx, cancel := snapshotCoordinatorContext(sess)
	20	+	done := quarantineSnapshotCoordinator(sess)
	21	+
	22	+	select {
	23	+	case <-ctx.Done():
	24	+	case <-time.After(testWaitTimeout):
	25	+	t.Fatal("quarantine did not cancel publication")
	26	+	}
	27	+	select {
	28	+	case <-done:
	29	+	case <-time.After(testWaitTimeout):
	30	+	t.Fatal("quarantine did not join idle coordinator")
	31	+	}
	32	+	cancel()
	33	+	}
	34	+
	35	+	func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
	36	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	37	+	repository := portsmocks.NewMockSnapshotRepository(t)
	38	+	WithSnapshotRepository(repository, nil)(d)
	39	+	startSnapshotEncodeWorker(t, d)
	40	+
	41	+	started := make(chan struct{})
	42	+	release := make(chan struct{})
	43	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.SnapshotPublication) error {
	44	+	close(started)
	45	+	<-release // deliberately ignore cancellation
	46	+	return nil
	47	+	}).Once()
	48	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	49	+	markSnapshotDirty(sess)
	50	+	require.True(t, d.scheduleSnapshot(sess))
	51	+	select {
	52	+	case <-started:
	53	+	case <-time.After(testWaitTimeout):
	54	+	t.Fatal("publication did not start")
	55	+	}
	56	+
	57	+	done := quarantineSnapshotCoordinator(sess)
	58	+	select {
	59	+	case <-done:
	60	+	t.Fatal("quarantine joined before publication returned")
	61	+	default:
	62	+	}
	63	+	close(release)
	64	+	select {
	65	+	case <-done:
	66	+	case <-time.After(testWaitTimeout):
	67	+	t.Fatal("quarantine did not join publication")
	68	+	}
	69	+	}
	70	+
17	71		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
18	72		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
19	73		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)


@@ -14,6 +14,60 @@ import (
14	portsmocks "github.com/bnema/vev/internal/ports/mocks"	14		portsmocks "github.com/bnema/vev/internal/ports/mocks"
15	)	15		)
16		16
		17	+	func TestSnapshotCoordinatorQuarantineCancelsPublicationBeforeDelete(t *testing.T) {
		18	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		19	+	ctx, cancel := snapshotCoordinatorContext(sess)
		20	+	done := quarantineSnapshotCoordinator(sess)
		21	+
		22	+	select {
		23	+	case <-ctx.Done():
		24	+	case <-time.After(testWaitTimeout):
		25	+	t.Fatal("quarantine did not cancel publication")
		26	+	}
		27	+	select {
		28	+	case <-done:
		29	+	case <-time.After(testWaitTimeout):
		30	+	t.Fatal("quarantine did not join idle coordinator")
		31	+	}
		32	+	cancel()
		33	+	}
		34	+
		35	+	func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
		36	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
		37	+	repository := portsmocks.NewMockSnapshotRepository(t)
		38	+	WithSnapshotRepository(repository, nil)(d)
		39	+	startSnapshotEncodeWorker(t, d)
		40	+
		41	+	started := make(chan struct{})
		42	+	release := make(chan struct{})
		43	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.SnapshotPublication) error {
		44	+	close(started)
		45	+	<-release // deliberately ignore cancellation
		46	+	return nil
		47	+	}).Once()
		48	+	sess := newSnapshotTestSession(t, "work", false, "/work")
		49	+	markSnapshotDirty(sess)
		50	+	require.True(t, d.scheduleSnapshot(sess))
		51	+	select {
		52	+	case <-started:
		53	+	case <-time.After(testWaitTimeout):
		54	+	t.Fatal("publication did not start")
		55	+	}
		56	+
		57	+	done := quarantineSnapshotCoordinator(sess)
		58	+	select {
		59	+	case <-done:
		60	+	t.Fatal("quarantine joined before publication returned")
		61	+	default:
		62	+	}
		63	+	close(release)
		64	+	select {
		65	+	case <-done:
		66	+	case <-time.After(testWaitTimeout):
		67	+	t.Fatal("quarantine did not join publication")
		68	+	}
		69	+	}
		70	+
17	func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {	71		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
18	clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}	72		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
19	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)	73		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)