maintenance.go RawHistoryFolder

1package daemon
2
3import (
4	"context"
5	"fmt"
6	"time"
7)
8
9// snapshotGarbageCollectionInterval spaces the periodic snapshot GC passes. A
10// long-running daemon otherwise accumulates superseded generations and objects
11// until its next restart, which then pays for the whole backlog.
12const snapshotGarbageCollectionInterval = 30 * time.Minute
13
14// startDurableMaintenance runs snapshot GC off the startup critical path: once
15// catalogue restoration has finished, then periodically. The recovery
16// coordinator fences each incarnation's step against catalogue mutations and
17// checkpoint publications, so a mutation waits for one incarnation at most.
18func (d *Daemon) startDurableMaintenance() {
19	if d == nil || d.recovery == nil || !d.snapshotGarbageCollection {
20		return
21	}
22	d.snapshotWorkerMu.Lock()
23	if d.maintenanceWorkerCancel != nil {
24		d.snapshotWorkerMu.Unlock()
25		return
26	}
27	ctx, cancel := context.WithCancel(d.serveCtx)
28	d.maintenanceWorkerCancel = cancel
29	done := make(chan struct{})
30	d.maintenanceWorkerDone = done
31	d.snapshotWorkerMu.Unlock()
32
33	go func() {
34		defer close(done)
35		d.runDurableMaintenance(ctx)
36	}()
37}
38
39// cancelDurableMaintenance stops the GC worker without waiting for it;
40// WaitDurableWriters owns the join.
41func (d *Daemon) cancelDurableMaintenance() {
42	d.snapshotWorkerMu.Lock()
43	defer d.snapshotWorkerMu.Unlock()
44	d.cancelDurableMaintenanceLocked()
45}
46
47// cancelDurableMaintenanceLocked requires snapshotWorkerMu.
48func (d *Daemon) cancelDurableMaintenanceLocked() {
49	if d.maintenanceWorkerCancel != nil {
50		d.maintenanceWorkerCancel()
51	}
52}
53
54func (d *Daemon) runDurableMaintenance(ctx context.Context) {
55	// Restoration reads committed checkpoints; collecting afterwards keeps the
56	// pass from competing with it for disk and for the coordinator fence.
57	select {
58	case <-ctx.Done():
59		return
60	case <-d.restoreDone:
61	}
62	for {
63		if err := d.collectSnapshotGarbage(ctx); err != nil && ctx.Err() == nil {
64			d.log.Warn("snapshot_garbage_collection_failed", "err", err)
65		}
66		timer := d.clock.NewTimer(snapshotGarbageCollectionInterval)
67		select {
68		case <-ctx.Done():
69			timer.Stop()
70			return
71		case <-timer.C():
72		}
73	}
74}
75
76// collectSnapshotGarbage runs one coordinator-owned GC pass.
77func (d *Daemon) collectSnapshotGarbage(ctx context.Context) error {
78	incarnations, err := d.recovery.CollectGarbage(ctx)
79	if err != nil {
80		return fmt.Errorf("collect snapshots: %w", err)
81	}
82	d.log.Info("snapshot_garbage_collection_complete", "incarnations", incarnations)
83	return nil
84}