orchard

  1package main
  2
  3import (
  4	"context"
  5	"database/sql"
  6	"fmt"
  7	"log/slog"
  8	"time"
  9)
 10
 11// Events age out after a year. The collector is a public unauthenticated POST
 12// and the collector id is in the page source of every tracked site, so without
 13// this the only bound on the table is how long somebody feels like posting.
 14
 15const (
 16	// A year, so the dashboard's longest window still has every row behind it.
 17	eventRetention = 365 * 24 * time.Hour
 18
 19	// Hourly rather than daily, so a deletion is always small and a machine
 20	// asleep most of the day still gets one in.
 21	sweepInterval = time.Hour
 22
 23	// Rows per DELETE. SQLite holds a database-wide write lock for the whole
 24	// statement, so an unbounded delete would stall every collect behind it.
 25	sweepChunk = 5000
 26
 27	// A ceiling on one table in one sweep, so a large backlog is worked off
 28	// over hours instead of in a single long lock.
 29	sweepMaxChunks = 200
 30
 31	// Reclaiming is paced the same way, about 4MB per step.
 32	vacuumPages    = 1000
 33	vacuumMaxSteps = 200
 34)
 35
 36// sweptTables are the two that grow with traffic. properties is operator sized
 37// and never swept.
 38var sweptTables = []string{"events", "bot_events"}
 39
 40type Sweeper struct {
 41	db *sql.DB
 42}
 43
 44func NewSweeper(db *sql.DB) *Sweeper { return &Sweeper{db: db} }
 45
 46// Run sweeps on a ticker until ctx is cancelled, and once at startup, since a
 47// process restarted more often than the interval would otherwise never sweep.
 48func (s *Sweeper) Run(ctx context.Context) {
 49	s.sweep(ctx)
 50
 51	t := time.NewTicker(sweepInterval)
 52	defer t.Stop()
 53	for {
 54		select {
 55		case <-ctx.Done():
 56			return
 57		case <-t.C:
 58			s.sweep(ctx)
 59		}
 60	}
 61}
 62
 63func (s *Sweeper) sweep(ctx context.Context) {
 64	cutoff := time.Now().Add(-eventRetention).UnixMilli()
 65	start := time.Now()
 66
 67	var total int64
 68	for _, table := range sweptTables {
 69		n, err := s.deleteOlderThan(ctx, table, cutoff)
 70		total += n
 71		if err != nil {
 72			slog.Error("retention sweep failed",
 73				slog.String("component", "retention"),
 74				slog.String("table", table),
 75				slog.Any("err", err))
 76			return
 77		}
 78	}
 79	if total == 0 {
 80		return
 81	}
 82
 83	slog.Info("retention sweep",
 84		slog.String("component", "retention"),
 85		slog.Int64("deleted", total),
 86		slog.String("cutoff", time.UnixMilli(cutoff).UTC().Format(time.RFC3339)),
 87		slog.String("reclaim", s.reclaim(ctx)),
 88		slog.Float64("ms", float64(time.Since(start).Microseconds())/1000))
 89}
 90
 91// deleteOlderThan removes aged rows from one table in bounded chunks. The
 92// subselect on rowid is there because LIMIT on DELETE is a SQLite compile-time
 93// option not enabled in every build.
 94func (s *Sweeper) deleteOlderThan(ctx context.Context, table string, cutoff int64) (int64, error) {
 95	// table is one of sweptTables above and never comes from a request, which
 96	// is the only reason it can be formatted into the statement at all.
 97	stmt := fmt.Sprintf(`
 98		DELETE FROM %s
 99		WHERE rowid IN (SELECT rowid FROM %s WHERE created_at < ? ORDER BY rowid LIMIT ?)`,
100		table, table)
101
102	var total int64
103	for i := 0; i < sweepMaxChunks; i++ {
104		res, err := s.db.ExecContext(ctx, stmt, cutoff, sweepChunk)
105		if err != nil {
106			return total, err
107		}
108		n, err := res.RowsAffected()
109		if err != nil {
110			return total, err
111		}
112		total += n
113		if n < sweepChunk {
114			return total, nil
115		}
116
117		// Yield, or a tight loop starves the collector for the whole sweep.
118		select {
119		case <-ctx.Done():
120			return total, ctx.Err()
121		case <-time.After(50 * time.Millisecond):
122		}
123	}
124	return total, nil
125}
126
127// reclaim hands freed pages back to the filesystem, and says what it did.
128//
129// Only a database created with auto_vacuum=INCREMENTAL can do this, and the
130// pragma is ignored on a file that already exists, so a database from before
131// that setting reports "not enabled" here. Deleting still bounds the row count
132// and stops the file growing, the freed pages are just reused rather than
133// returned. A one-off `VACUUM` converts an existing file if the space is wanted
134// back sooner.
135func (s *Sweeper) reclaim(ctx context.Context) string {
136	var mode int
137	if err := s.db.QueryRowContext(ctx, "PRAGMA auto_vacuum").Scan(&mode); err != nil {
138		return fmt.Sprintf("auto_vacuum check failed: %v", err)
139	}
140	if mode != 2 {
141		return "not enabled"
142	}
143
144	for i := 0; i < vacuumMaxSteps; i++ {
145		// The page count is formatted in because SQLite takes no PRAGMA
146		// parameter. A bare incremental_vacuum would free the whole freelist
147		// under the write lock, hence the pacing.
148		if _, err := s.db.ExecContext(ctx,
149			fmt.Sprintf("PRAGMA incremental_vacuum(%d)", vacuumPages)); err != nil {
150			return fmt.Sprintf("incremental_vacuum failed: %v", err)
151		}
152
153		var remaining int64
154		if err := s.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&remaining); err != nil {
155			return fmt.Sprintf("freelist_count failed: %v", err)
156		}
157		if remaining == 0 {
158			break
159		}
160
161		select {
162		case <-ctx.Done():
163			return "cancelled"
164		case <-time.After(50 * time.Millisecond):
165		}
166	}
167	return "reclaimed"
168}