1package daemon
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sort"
8 "strings"
9 "sync"
10
11 "github.com/bnema/vev/internal/domain"
12)
13
14// errPurgeAdmissionClosed is returned to command-path creation/resume that
15// already holds architecture locks and therefore cannot wait for a daemon-wide
16// purge. Route-level creation waits on the purge admission gate instead.
17var errPurgeAdmissionClosed = errors.New("daemon is purging sessions")
18
19// purgeAdmissionClosedLocked reports whether a KillAll purge currently owns
20// admission. Caller holds mu. A command-path create/resume that cannot wait
21// rejects with errPurgeAdmissionClosed instead of inserting into the exact set
22// being removed.
23func (d *Daemon) purgeAdmissionClosedLocked() bool { return d.purgeAdmissionClosing }
24
25// acquirePurgeAdmission reserves one create/restore transition against the
26// transient KillAll gate. It waits without holding any other lock while a purge
27// owns the gate, so a Hello arriving mid-purge is admitted as a create or
28// restore after the purge rather than inserted into the set being removed.
29func (d *Daemon) acquirePurgeAdmission(ctx context.Context) error {
30 for {
31 d.mu.Lock()
32 if !d.purgeAdmissionClosing {
33 d.purgeAdmissionActive++
34 d.mu.Unlock()
35 return nil
36 }
37 changed := d.purgeAdmissionChangeLocked()
38 d.mu.Unlock()
39 select {
40 case <-changed:
41 case <-ctx.Done():
42 return ctx.Err()
43 }
44 }
45}
46
47// releasePurgeAdmission drops a reservation taken by acquirePurgeAdmission or
48// an equivalent move reservation.
49func (d *Daemon) releasePurgeAdmission() {
50 d.mu.Lock()
51 d.purgeAdmissionActive--
52 d.signalPurgeAdmissionChangedLocked()
53 d.mu.Unlock()
54}
55
56// purgeAdmissionStatus reports how a KillAll's attempt to close the transient
57// admission gate resolved.
58type purgeAdmissionStatus uint8
59
60const (
61 // purgeAdmissionGranted means the gate closed and every admitted transition
62 // drained within the bound, so the purge owns the exact lifecycle set.
63 purgeAdmissionGranted purgeAdmissionStatus = iota
64 // purgeAdmissionSuperseded means an explicit daemon shutdown began, or the
65 // caller's context ended, before the drain completed. The purge removes
66 // nothing and defers to the stop.
67 purgeAdmissionSuperseded
68 // purgeAdmissionTimedOut means an admitted transition did not release the
69 // gate before the bounded purge budget expired. The purge removes nothing so
70 // a transition stuck in repository I/O that ignores cancellation can never
71 // wedge Serve.
72 purgeAdmissionTimedOut
73)
74
75// beginPurgeAdmission closes the KillAll admission gate and waits, bounded, for
76// every admitted transition to drain before the caller snapshots the exact
77// lifecycle set. It mirrors closeMoveLifecycles but is re-openable: the caller
78// must always call endPurgeAdmission to clear it and leave the daemon serving.
79//
80// The wait ends as soon as the drain completes. It also ends early, without the
81// purge owning anything, when ctx is done, an explicit shutdown begins
82// (closing), or deadline expires — so an admitted create, restore, or move that
83// ignores cancellation (for example a repository call that never returns) can
84// never extend Serve past the bounded purge budget. On success purgeEpoch is
85// incremented so a restoration worker that waited on the gate can detect that
86// the record set it captured was superseded.
87func (d *Daemon) beginPurgeAdmission(ctx context.Context, deadline *snapshotShutdownDeadline) purgeAdmissionStatus {
88 if ctx == nil {
89 ctx = context.Background()
90 }
91 d.mu.Lock()
92 d.purgeAdmissionClosing = true
93 d.signalPurgeAdmissionChangedLocked()
94 for {
95 if d.closing {
96 d.mu.Unlock()
97 return purgeAdmissionSuperseded
98 }
99 if d.purgeAdmissionActive == 0 {
100 d.purgeEpoch++
101 d.signalPurgeAdmissionChangedLocked()
102 d.mu.Unlock()
103 return purgeAdmissionGranted
104 }
105 changed := d.purgeAdmissionChangeLocked()
106 d.mu.Unlock()
107 select {
108 case <-changed:
109 case <-ctx.Done():
110 return purgeAdmissionSuperseded
111 case <-deadline.Done():
112 // An explicit stop that began as the budget expired wins: classify it
113 // as superseded so the purge defers and reports stop semantics instead
114 // of a drain timeout. terminateAllForShutdown also signals changed, but
115 // a ready select may still pick the deadline, so recheck closing here.
116 d.mu.Lock()
117 closing := d.closing
118 d.mu.Unlock()
119 if closing {
120 return purgeAdmissionSuperseded
121 }
122 return purgeAdmissionTimedOut
123 }
124 d.mu.Lock()
125 }
126}
127
128func (d *Daemon) endPurgeAdmission() {
129 d.mu.Lock()
130 d.purgeAdmissionClosing = false
131 d.signalPurgeAdmissionChangedLocked()
132 d.mu.Unlock()
133}
134
135// purgeEpochSnapshot returns the current purge boundary. Restoration captures
136// it before reading catalogue records so a purge that completes while its
137// workers wait for admission invalidates those records.
138func (d *Daemon) purgeEpochSnapshot() uint64 {
139 d.mu.Lock()
140 defer d.mu.Unlock()
141 return d.purgeEpoch
142}
143
144// purgeSupersededLocked reports whether an admitted transition captured at
145// epoch must abandon its work: an explicit shutdown has begun, a purge currently
146// owns and waits on admission, or a purge completed after epoch was captured.
147// Caller holds mu.
148func (d *Daemon) purgeSupersededLocked(epoch uint64) bool {
149 return d.closing || d.purgeAdmissionClosing || d.purgeEpoch != epoch
150}
151
152// purgeSuperseded is purgeSupersededLocked without the caller-held lock.
153func (d *Daemon) purgeSuperseded(epoch uint64) bool {
154 d.mu.Lock()
155 defer d.mu.Unlock()
156 return d.purgeSupersededLocked(epoch)
157}
158
159// purgeSupersededForShutdown reports whether an explicit daemon shutdown has
160// begun. A KillAll purge defers to it so explicit stop semantics (preserving
161// live sessions as stopped authority) are never overwritten by a concurrent
162// purge.
163func (d *Daemon) purgeSupersededForShutdown() bool {
164 d.mu.Lock()
165 defer d.mu.Unlock()
166 return d.closing
167}
168
169// purgeAdmissionChangeLocked returns the channel closed when either purge
170// admission field changes. Caller holds mu.
171func (d *Daemon) purgeAdmissionChangeLocked() chan struct{} {
172 if d.purgeAdmissionChanged == nil {
173 d.purgeAdmissionChanged = make(chan struct{})
174 }
175 return d.purgeAdmissionChanged
176}
177
178// signalPurgeAdmissionChangedLocked wakes every admission waiter. Caller holds
179// mu.
180func (d *Daemon) signalPurgeAdmissionChangedLocked() {
181 close(d.purgeAdmissionChangeLocked())
182 d.purgeAdmissionChanged = make(chan struct{})
183}
184
185// purgeRecordClass identifies which lifecycle record class a failed purge unit
186// belongs to, so callers can report a typed partial failure instead of one
187// opaque error.
188type purgeRecordClass uint8
189
190const (
191 purgeRecordLive purgeRecordClass = iota + 1
192 purgeRecordStopped
193 purgeRecordBroken
194 purgeRecordDurable
195)
196
197func (c purgeRecordClass) String() string {
198 switch c {
199 case purgeRecordLive:
200 return "live"
201 case purgeRecordStopped:
202 return "stopped"
203 case purgeRecordBroken:
204 return "broken"
205 case purgeRecordDurable:
206 return "durable"
207 default:
208 return "unknown"
209 }
210}
211
212// inactivePurgeClass resolves the record class of a captured stopped/broken
213// registry entry.
214func inactivePurgeClass(entry inactiveSession) purgeRecordClass {
215 if entry.broken() {
216 return purgeRecordBroken
217 }
218 return purgeRecordStopped
219}
220
221// purgeFailure is one structured partial purge failure: the record class, the
222// affected name when known, and the underlying error.
223type purgeFailure struct {
224 Class purgeRecordClass
225 Name string
226 Err error
227}
228
229func (f purgeFailure) Error() string {
230 if f.Name == "" {
231 return fmt.Sprintf("%s: %v", f.Class, f.Err)
232 }
233 return fmt.Sprintf("%s %q: %v", f.Class, f.Name, f.Err)
234}
235
236func (f purgeFailure) Unwrap() error { return f.Err }
237
238// purgeResult reports every record that could not be fully purged. An empty
239// Failures slice means the purge removed the entire exact lifecycle set. A
240// superseded result means an explicit daemon shutdown won the race: the purge
241// deferred and removed nothing, or removed only the units whose teardown it had
242// already owned when the stop began. A drain timeout result means an admitted
243// transition refused to drain within the bounded admission budget, so the purge
244// removed nothing and reopened admission. A typed live failure means a live unit
245// was left untouched because the shared live-teardown budget expired before its
246// destructive boundary or because its participants kept changing. A typed
247// stopped/broken failure means the bounded stopped/broken sweep did not finish
248// the record's durable purge within its own budget.
249type purgeResult struct {
250 Failures []purgeFailure
251 superseded bool
252 drainTimedOut bool
253}
254
255func (r purgeResult) Failed() bool { return len(r.Failures) > 0 || r.drainTimedOut }
256
257// Superseded reports that an explicit daemon shutdown owned the lifecycle set
258// before this purge could finish destroying it, so the purge deferred to it.
259// Any Failures it carries describe units the stop still owns; a superseded
260// result never means the purge silently reported success.
261func (r purgeResult) Superseded() bool { return r.superseded }
262
263// purgeSummaryFailureLimit bounds the wire-safe failure detail. The result may
264// hold arbitrarily many failures (one per record), but a control response must
265// stay bounded: report the total count and only the first N failures, then note
266// how many were omitted.
267const purgeSummaryFailureLimit = 5
268
269func (r purgeResult) summary() string {
270 if r.drainTimedOut {
271 return "session purge aborted: purge admission did not drain within the bounded timeout; no records were removed"
272 }
273 if !r.Failed() {
274 return ""
275 }
276 shown := min(len(r.Failures), purgeSummaryFailureLimit)
277 parts := make([]string, 0, shown)
278 for _, failure := range r.Failures[:shown] {
279 parts = append(parts, failure.Error())
280 }
281 if omitted := len(r.Failures) - shown; omitted > 0 {
282 return fmt.Sprintf("%d record(s) could not be purged; first %d: %s; %d more omitted", len(r.Failures), shown, strings.Join(parts, "; "), omitted)
283 }
284 return fmt.Sprintf("%d record(s) could not be purged: %s", len(r.Failures), strings.Join(parts, "; "))
285}
286
287// purgeAllSessions is the KillAll operation: it removes every live session and
288// every stopped/broken/durable record while leaving the daemon active. It does
289// not set closing, close move admission, cancel daemon-wide pane/process
290// contexts, close done, or consume shutdown signals, and it never makes Serve
291// return.
292//
293// Linearization: beginPurgeAdmission closes the transient admission gate and
294// drains in-flight create/restore/move reservations, so the exact lifecycle set
295// captured below cannot be mutated by a concurrent transition. Purge then runs
296// against that exact set, and endPurgeAdmission reopens admission for later
297// creations, which are ordered after the purge.
298//
299// Budgets: the admission drain, the stopped/broken durable sweep, and
300// live-session teardown spend separate bounded budgets, each matching the
301// daemon shutdown budget. A transition that ignores
302// cancellation and burns the whole drain budget therefore cannot leave teardown
303// with no time, and teardown cannot extend the drain. A repository deletion
304// that ignores cancellation cannot extend the stopped/broken sweep past its own
305// budget either. The live budget is armed
306// when the live phase begins and is shared by every live unit: once it expires,
307// each remaining live unit is skipped and reported as a typed live failure
308// rather than reported as purged. Every unit keeps its own accurate name.
309func (d *Daemon) purgeAllSessions(reason uint8) purgeResult {
310 d.purgeAllMu.Lock()
311 defer d.purgeAllMu.Unlock()
312
313 // One deadline bounds the drain of every admitted transition. It is stopped
314 // before the destructive phase so teardown spends a fresh budget.
315 admissionDeadline := newSnapshotShutdownDeadline(d.clock)
316 defer admissionDeadline.stop()
317
318 ctx := d.serveCtx
319 if ctx == nil {
320 ctx = context.Background()
321 }
322 switch d.beginPurgeAdmission(ctx, admissionDeadline) {
323 case purgeAdmissionSuperseded:
324 d.log.Info("session purge superseded before admission", "reason", reason)
325 d.endPurgeAdmission()
326 return purgeResult{superseded: true}
327 case purgeAdmissionTimedOut:
328 d.log.Warn("session purge aborted: purge admission did not drain", "reason", reason)
329 d.endPurgeAdmission()
330 return purgeResult{drainTimedOut: true}
331 }
332 admissionDeadline.stop()
333 defer d.endPurgeAdmission()
334
335 d.mu.Lock()
336 // A shutdown that began after admission was granted still wins: no record
337 // has been removed yet, so defer without touching the preserved set.
338 if d.closing {
339 d.mu.Unlock()
340 d.log.Info("session purge superseded by daemon shutdown", "reason", reason, "phase", "pre-snapshot")
341 return purgeResult{superseded: true}
342 }
343 d.purgeAllParkingLocked()
344 parkedRetirements := d.purgeAllParkedLocked()
345 d.purgeAllSuspendedLocked()
346 live := d.sessionsSnapshotLocked()
347 inactive := make([]inactiveSession, 0, len(d.inactive))
348 exact := make(map[string]struct{}, len(live)+len(d.inactive))
349 for _, entry := range d.inactive {
350 inactive = append(inactive, entry)
351 exact[entry.name] = struct{}{}
352 }
353 for _, sess := range live {
354 sess.mu.Lock()
355 exact[sess.name] = struct{}{}
356 sess.mu.Unlock()
357 }
358 d.mu.Unlock()
359 if d.afterPurgeAdmission != nil {
360 d.afterPurgeAdmission()
361 }
362 d.finishParkedAttachmentRetirements(parkedRetirements)
363 sort.Slice(inactive, func(i, j int) bool { return inactive[i].name < inactive[j].name })
364
365 // A shutdown that began while the snapshot was taken still wins: no record
366 // has been removed yet, so defer without touching the preserved set.
367 if d.purgeSupersededForShutdown() {
368 d.log.Info("session purge superseded by daemon shutdown", "reason", reason, "phase", "pre-destructive")
369 return purgeResult{superseded: true}
370 }
371
372 result := purgeResult{}
373 d.log.Info("session purge begin", "reason", reason, "live_sessions", len(live), "inactive_sessions", len(inactive))
374
375 // Stopped authority first (mirroring the shutdown purge order), then live
376 // sessions. Each unit is independent: one failure never cancels a sibling.
377 // Explicit stop wins per unit: a shutdown that begins mid-purge stops further
378 // removals so it can preserve every not-yet-purged lifecycle as stopped
379 // authority. A unit the purge already owns (its teardown is in progress)
380 // keeps purge authority and completes as a purge; only units not yet started
381 // are handed back to the stop.
382 //
383 // The stopped/broken sweep spends its own deadline and context, separate from
384 // the admission drain and the live-teardown budget. A repository deletion
385 // that ignores cancellation therefore cannot extend the purge: at the budget
386 // every record the sweep did not finish is still stopped/broken authority and
387 // is reported as a typed failure, and admission is released promptly.
388 if len(inactive) > 0 {
389 stoppedDeadline := newSnapshotShutdownDeadline(d.clock)
390 defer stoppedDeadline.stop()
391 stoppedCtx, cancelStopped := snapshotStopContext(stoppedDeadline)
392 defer cancelStopped()
393 sweep := d.purgeInactiveSweep(inactive, stoppedCtx, stoppedDeadline)
394 result.Failures = append(result.Failures, sweep.failures...)
395 if sweep.superseded {
396 d.log.Info("session purge superseded by daemon shutdown", "reason", reason, "phase", "stopped")
397 return purgeResult{Failures: result.Failures, superseded: true}
398 }
399 }
400 // The live phase gets its own bounded budget, armed here so neither the
401 // admission drain nor the stopped-record purge consumes it. It is shared by
402 // every live unit.
403 destructiveDeadline := newSnapshotShutdownDeadline(d.clock)
404 defer destructiveDeadline.stop()
405 for _, sess := range live {
406 if d.purgeSupersededForShutdown() {
407 d.log.Info("session purge superseded by daemon shutdown", "reason", reason, "phase", "live")
408 return purgeResult{Failures: result.Failures, superseded: true}
409 }
410 sess.mu.Lock()
411 name := sess.name
412 sess.mu.Unlock()
413 if d.beforePurgeLiveSessionKill != nil {
414 d.beforePurgeLiveSessionKill(sess)
415 }
416 sess.stopInMemoryLifecycle()
417 err := d.purgeLiveSession(sess, reason, destructiveDeadline)
418 if err == nil {
419 // A stop that wins this unit after the precheck above makes
420 // purgeLiveSession a no-op: the stop already removed the session and
421 // now owns it. Recheck so the purge never counts a stop-owned unit as a
422 // successful purge. A unit whose teardown the purge already owned keeps
423 // purge authority and is reported as removed, but the remaining units
424 // still belong to the stop.
425 if d.purgeSupersededForShutdown() {
426 d.log.Info("session purge superseded by daemon shutdown", "reason", reason, "phase", "live-owned")
427 return purgeResult{Failures: result.Failures, superseded: true}
428 }
429 continue
430 }
431 if d.purgeSupersededForShutdown() &&
432 (errors.Is(err, errSessionKillDeadline) || errors.Is(err, errSessionKillParticipantsChanged)) {
433 // The explicit stop owns the remaining lifecycles: either the live budget
434 // expired as it began, or a participant change aborted a kill that the
435 // stop now classifies as preservation rather than a purge failure.
436 d.log.Info("session purge superseded by daemon shutdown", "reason", reason, "phase", "live-deadline")
437 return purgeResult{Failures: result.Failures, superseded: true}
438 }
439 result.Failures = append(result.Failures, purgeFailure{Class: purgeRecordLive, Name: name, Err: err})
440 d.log.Error("purging live session failed", "err", err, "session", name)
441 }
442 result.Failures = append(result.Failures, d.purgeDurableResidue(exact, result.Failures)...)
443 sort.SliceStable(result.Failures, func(i, j int) bool {
444 if result.Failures[i].Class != result.Failures[j].Class {
445 return result.Failures[i].Class < result.Failures[j].Class
446 }
447 return result.Failures[i].Name < result.Failures[j].Name
448 })
449 if result.Failed() {
450 d.log.Warn("session purge incomplete", "reason", reason, "failures", len(result.Failures))
451 }
452 return result
453}
454
455// purgeDurableResidue removes durable records that have no live/stopped/broken
456// registry entry, so a KillAll still accounts for every catalogue record. A
457// record already represented by a reported live/stopped/broken failure is that
458// failure's durable residue and is not double-reported here.
459func (d *Daemon) purgeDurableResidue(exact map[string]struct{}, existing []purgeFailure) []purgeFailure {
460 if !d.persistEnabled || d.catalogue == nil {
461 return nil
462 }
463 records, err := d.catalogue.Records()
464 if err != nil {
465 return []purgeFailure{{Class: purgeRecordDurable, Err: err}}
466 }
467 reported := make(map[string]struct{}, len(existing))
468 for _, failure := range existing {
469 reported[failure.Name] = struct{}{}
470 }
471 var out []purgeFailure
472 for _, record := range records {
473 if _, accounted := exact[record.Name]; accounted {
474 continue
475 }
476 if _, already := reported[record.Name]; already {
477 continue
478 }
479 if err := d.purgeDurableRecord(record); err != nil {
480 out = append(out, purgeFailure{Class: purgeRecordDurable, Name: record.Name, Err: err})
481 }
482 }
483 return out
484}
485
486// errSessionPurgeSweepDeadline reports that the stopped/broken sweep's own
487// bounded budget expired before a record's durable purge finished. The record
488// keeps its stopped/broken authority and is reported as a typed failure rather
489// than silently counted as purged.
490var errSessionPurgeSweepDeadline = errors.New("stopped session purge sweep deadline exceeded")
491
492// errSessionPurgeDeletePending reports that a purge-owned live unit's exact
493// durable delete did not finish within the shared live-teardown budget. The
494// live identity is already removed and its stopped authority still reserves the
495// name; a detached worker finishes the fenced DeleteExact and removes that
496// authority once the repository returns. It is a typed failure so a bounded
497// purge never reports a skipped durable delete as success, and a repository
498// delete that ignores cancellation can never hold purgeAllMu, the admission
499// gate, or control responses.
500var errSessionPurgeDeletePending = errors.New("session durable purge pending")
501
502// finishLiveDurablePurge runs the exact durable delete for a live unit the purge
503// already removed from the live registry. With a bounded deadline it runs the
504// delete on a detached worker: the caller returns errSessionPurgeDeletePending
505// at the deadline while the worker finishes and removes the matching stopped
506// authority. That removal is fenced by the captured name, incarnation, and
507// createdAt, so a same-name or re-incarnated replacement is never touched. A
508// nil deadline (a command-path kill) keeps the previous synchronous behavior.
509func (d *Daemon) finishLiveDurablePurge(deadline *snapshotShutdownDeadline, record domain.CatalogueRecord) error {
510 expected := inactiveSession{name: record.Name, createdAt: record.CreatedAt, incarnation: record.IncarnationID, purging: true}
511 finish := func() error {
512 if err := d.finishSnapshotPurge(d.serveCtx, record.Name, record.IncarnationID, record.CreatedAt); err != nil {
513 return err
514 }
515 d.mu.Lock()
516 if stopped, ok := d.inactive[record.Name]; ok && stopped.purging && stopped.sameLifecycle(expected) {
517 delete(d.inactive, record.Name)
518 }
519 d.mu.Unlock()
520 d.reconcileAllRouteHistories()
521 return nil
522 }
523 if deadline == nil {
524 return finish()
525 }
526 done := make(chan error, 1)
527 go func() { done <- finish() }()
528 select {
529 case err := <-done:
530 return err
531 case <-deadline.Done():
532 // Recheck completion so a delete that finished as the budget expired is
533 // reported as success rather than pending.
534 select {
535 case err := <-done:
536 return err
537 default:
538 // The caller already reported a typed pending failure and stopped
539 // waiting. Wait for the detached delete on its own goroutine so a late
540 // failure is still recorded with bounded, non-sensitive fields (record
541 // class, incarnation, created-time) instead of the user-chosen session
542 // name.
543 go func() {
544 if err := <-done; err != nil {
545 d.log.Warn("detached session purge delete failed after deadline",
546 "class", purgeRecordLive, "incarnation", record.IncarnationID, "created_at", record.CreatedAt, "err", err)
547 }
548 }()
549 return errSessionPurgeDeletePending
550 }
551 }
552}
553
554// purgeInactiveSweepResult reports the bounded stopped/broken sweep outcome.
555// failures holds every record the purge could not finish; superseded means an
556// explicit stop began mid-sweep and owns the remaining records.
557type purgeInactiveSweepResult struct {
558 failures []purgeFailure
559 superseded bool
560}
561
562// purgeInactiveSweep removes the captured stopped/broken records under its own
563// deadline and context, separate from the admission drain and the
564// live-teardown budget. Because a repository deletion may ignore cancellation,
565// the sweep runs on a detached worker and the caller returns at the deadline:
566// every record the worker did not finish keeps its stopped/broken authority and
567// is reported as a typed failure, so a bounded purge releases admission
568// promptly instead of waiting out an uncooperative delete. The detached worker
569// stops contributing to the returned result at that point; its registry removal
570// stays lifecycle-fenced, so a late completion cannot delete a same-name
571// replacement.
572func (d *Daemon) purgeInactiveSweep(inactive []inactiveSession, ctx context.Context, deadline *snapshotShutdownDeadline) purgeInactiveSweepResult {
573 var (
574 mu sync.Mutex
575 failures []purgeFailure
576 completed int
577 superseded bool
578 detached bool
579 )
580 sweepDone := make(chan struct{})
581 go func() {
582 defer close(sweepDone)
583 for _, entry := range inactive {
584 mu.Lock()
585 abandoned := detached
586 mu.Unlock()
587 if abandoned {
588 return
589 }
590 if d.purgeSupersededForShutdown() {
591 mu.Lock()
592 superseded = true
593 mu.Unlock()
594 return
595 }
596 class := inactivePurgeClass(entry)
597 err := d.retryStoppedPurgeContextExact(ctx, entry.name, entry.incarnation, &entry.createdAt)
598 mu.Lock()
599 if detached {
600 // The bounded caller already reported this unit; a late result must
601 // not mutate the returned result.
602 mu.Unlock()
603 return
604 }
605 if err != nil {
606 failures = append(failures, purgeFailure{Class: class, Name: entry.name, Err: err})
607 d.log.Error("purging stopped session failed", "err", err, "session", entry.name)
608 }
609 completed++
610 mu.Unlock()
611 }
612 }()
613
614 select {
615 case <-sweepDone:
616 case <-deadline.Done():
617 mu.Lock()
618 detached = true
619 done := completed
620 reported := append([]purgeFailure(nil), failures...)
621 mu.Unlock()
622 if d.purgeSupersededForShutdown() {
623 // An explicit stop that began as the budget expired owns the remaining
624 // records, so the sweep is supersession rather than a purge failure.
625 return purgeInactiveSweepResult{failures: reported, superseded: true}
626 }
627 for _, entry := range inactive[done:] {
628 reported = append(reported, purgeFailure{Class: inactivePurgeClass(entry), Name: entry.name, Err: errSessionPurgeSweepDeadline})
629 d.log.Error("purging stopped session failed", "err", errSessionPurgeSweepDeadline, "session", entry.name)
630 }
631 return purgeInactiveSweepResult{failures: reported}
632 }
633
634 mu.Lock()
635 defer mu.Unlock()
636 return purgeInactiveSweepResult{failures: failures, superseded: superseded}
637}
638
639// purgeLiveSession removes one live session under the shared live-teardown
640// budget. A post-freeze participant change aborts the kill before it touches the
641// session. A session a racing path already removed owns nothing and counts as
642// purged; an explicit stop that began meanwhile owns the rest and is never
643// retried against; otherwise the purge retries once against a fresh snapshot,
644// and a session whose participants changed again stays a typed live failure for
645// the caller.
646func (d *Daemon) purgeLiveSession(sess *session, reason uint8, deadline *snapshotShutdownDeadline) error {
647 err := d.killSessionWithSnapshotDeadline(sess, reason, true, deadline, nil)
648 if !errors.Is(err, errSessionKillParticipantsChanged) {
649 return err
650 }
651 if d.purgeSupersededForShutdown() {
652 // An explicit stop has begun: it owns the session's remaining lifecycle, so
653 // the purge must not retry against it. The caller classifies the abort as
654 // supersession instead of a live failure.
655 return err
656 }
657 if !d.sessionRegistered(sess) {
658 return nil
659 }
660 return d.killSessionWithSnapshotDeadline(sess, reason, true, deadline, nil)
661}
662
663// sessionRegistered reports whether sess is still the exact live registry entry.
664func (d *Daemon) sessionRegistered(sess *session) bool {
665 if sess == nil {
666 return false
667 }
668 d.mu.Lock()
669 defer d.mu.Unlock()
670 return d.sessions[sess.id] == sess
671}
672
673func (d *Daemon) purgeDurableRecord(record domain.CatalogueRecord) error {
674 if !d.persistEnabled {
675 return nil
676 }
677 if d.recovery == nil {
678 return errors.New("durable session authority is not configured")
679 }
680 return d.finishSnapshotPurge(d.serveCtx, record.Name, record.IncarnationID, record.CreatedAt)
681}