fix: Artemis drift at source

.env.example

@@ -32,6 +32,8 @@ VALKEY_ADDR=localhost:6379
 # ALIAS_PRODUCTION_KEY_FORMAT=<site>/production
 # ALIAS_PREVIEW_KEY_FORMAT=<site>/preview
 # DEPLOY_PREFIX_FORMAT=<site>/deploys/<ts>-<sha>/
+# PUBLIC_URL_PRODUCTION_FORMAT=https://<site>.freecode.camp
+# PUBLIC_URL_PREVIEW_FORMAT=https://<site>.preview.freecode.camp
 # UPLOAD_MAX_BYTES=104857600          # 100 MiB
 # LOG_LEVEL=info                      # debug | info | warn | error
 
@@ -63,7 +65,7 @@ VALKEY_ADDR=localhost:6379
 # CLEANUP_RETENTION_DAYS=7            # days before a superseded deploy is GC-eligible
 # CLEANUP_RECENT_KEEP=3               # newest N deploys per site always kept
 # CLEANUP_GRACE=72h                   # min deploy age before GC; must be >= JWT_TTL_SECONDS
-# CLEANUP_BLAST_CAP=0                 # max deletes per sweep; 0 disables the cap
+# CLEANUP_BLAST_CAP=10                # max deletes per sweep, oldest first; 0 refuses every destructive repair
 # CLEANUP_TRASH_PREFIX=_trash/        # R2 prefix for tombstoned objects
 # CLEANUP_RECOVERY_DAYS=7             # days a tombstone survives before hard purge
 # CLEANUP_DRY_RUN=                    # 1/true: plan-only, execute nothing

cmd/artemis/bootrun_test.go

@@ -52,8 +52,8 @@ func bootCfg(t *testing.T, dsn, valkeyAddr string, port int) *config.Config {
 	cfg.GitHub.MembershipCacheTTL = time.Minute
 	cfg.JWT.SigningKey = "0123456789abcdef0123456789abcdef"
 	cfg.JWT.TTL = 15 * time.Minute
-	cfg.Aliases.ProductionKeyFormat = "<site>/production"
-	cfg.Aliases.PreviewKeyFormat = "<site>/preview"
+	cfg.Aliases.ProductionKeyFormat = "<site>.example.test/production"
+	cfg.Aliases.PreviewKeyFormat = "<site>.example.test/preview"
 	cfg.Cleanup.TrashPrefix = "_trash/"
 	cfg.Cleanup.RecoveryDays = 7
 	cfg.Cleanup.Grace = 72 * time.Hour
@@ -332,6 +332,8 @@ func TestRun_BootsFromEnvAndExitsOnSigterm(t *testing.T) {
 	t.Setenv("GH_CLIENT_ID", "cid")
 	t.Setenv("JWT_SIGNING_KEY", "0123456789abcdef0123456789abcdef")
 	t.Setenv("DEPLOY_PREFIX_FORMAT", "<site>.example.test/deploys/<ts>-<sha>/")
+	t.Setenv("ALIAS_PRODUCTION_KEY_FORMAT", "<site>.example.test/production")
+	t.Setenv("ALIAS_PREVIEW_KEY_FORMAT", "<site>.example.test/preview")
 	t.Setenv("LOG_LEVEL", "error")
 	t.Setenv("SENTRY_DSN", "https://publickey@o0.ingest.sentry.io/0")
 	t.Setenv("ENVIRONMENT", "test")

cmd/artemis/dispatch_test.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDispatchSubcommand_NoArgsRunsTheServer(t *testing.T) {
+	t.Parallel()
+
+	handled, err := dispatchSubcommand(context.Background(), &bytes.Buffer{}, nil)
+	require.NoError(t, err)
+	require.False(t, handled, "an argv-less invocation is the server, not a subcommand")
+}
+
+func TestDispatchSubcommand_RejectsAnUnknownSubcommand(t *testing.T) {
+	t.Parallel()
+
+	handled, err := dispatchSubcommand(context.Background(), &bytes.Buffer{}, []string{"drift-report"})
+	require.True(t, handled,
+		"an unrecognised subcommand must not fall through to the server: a typo would boot a second "+
+			"artemis instead of reporting the typo")
+	require.ErrorContains(t, err, "drift-report")
+}
+
+func TestDispatchSubcommand_RejectsArgumentsToDriftReport(t *testing.T) {
+	t.Parallel()
+
+	handled, err := dispatchSubcommand(context.Background(), &bytes.Buffer{}, []string{driftReportCommand, "www"})
+	require.True(t, handled)
+	require.ErrorContains(t, err, "takes no arguments",
+		"driftreport swept every site regardless of argv, so a site name an operator typed was silently "+
+			"ignored and the whole-fleet report read as scoped")
+}

cmd/artemis/driftalert.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"sort"
 	"strings"
 )
 
@@ -13,8 +14,11 @@ const (
 	opDriftSelfCheck      = "drift.selfcheck"
 	opDriftUnreadable     = "drift.unreadable"
 	opDriftAliasedMissing = "drift.aliased_missing"
+	opDriftReclaimable    = "drift.reclaimable"
 )
 
+const reclaimableAlertThreshold = 25
+
 type driftVerdict struct {
 	Op    string
 	Err   error
@@ -38,6 +42,16 @@ func classifyDrift(res sweepResult) driftVerdict {
 	if unread != nil {
 		return driftVerdict{Op: opDriftUnreadable, Err: unread, Fails: true}
 	}
+	if reindex, tombstone, _, _ := res.totals(); reindex+tombstone >= reclaimableAlertThreshold {
+		sites := reclaimableSites(res.Reports)
+		return driftVerdict{
+			Op: opDriftReclaimable,
+			Err: fmt.Errorf(
+				"%d deploys are reclaimable across %s (>= %d): storage is accruing faster than it is "+
+					"collected; run `artemis reconcile <site> --apply` for each and find what stopped expiring",
+				reindex+tombstone, strings.Join(sites, ", "), reclaimableAlertThreshold),
+		}
+	}
 	return driftVerdict{}
 }
 
@@ -49,6 +63,30 @@ func unreadableErr(unreadable []string, sites int) error {
 		len(unreadable), sites, strings.Join(unreadable, ", "))
 }
 
+func reclaimableSites(reports []siteDrift) []string {
+	type sited struct {
+		site string
+		n    int
+	}
+	var ranked []sited
+	for _, r := range reports {
+		if n := len(r.Reindex) + len(r.Tombstone); n > 0 {
+			ranked = append(ranked, sited{r.Site, n})
+		}
+	}
+	sort.Slice(ranked, func(i, j int) bool {
+		if ranked[i].n != ranked[j].n {
+			return ranked[i].n > ranked[j].n
+		}
+		return ranked[i].site < ranked[j].site
+	})
+	out := make([]string, 0, len(ranked))
+	for _, s := range ranked {
+		out = append(out, fmt.Sprintf("%s (%d)", s.site, s.n))
+	}
+	return out
+}
+
 func unreadableSites(reports []siteDrift) []string {
 	var out []string
 	for _, r := range reports {

cmd/artemis/driftalert_threshold_test.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/freeCodeCamp/artemis/internal/observability"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func sweepWithReclaimable(reindex, tombstone int) sweepResult {
+	var s siteDrift
+	s.Site = "www.freecode.camp"
+	for i := 0; i < reindex; i++ {
+		s.Reindex = append(s.Reindex, "r")
+	}
+	for i := 0; i < tombstone; i++ {
+		s.Tombstone = append(s.Tombstone, "t")
+	}
+	return sweepResult{
+		Reports: []siteDrift{s},
+		Stats:   sweepStats{Sites: 1, PGDeploys: 1, IndexedTotal: 1, R2Objects: 1},
+	}
+}
+
+func TestClassifyDrift_StaysQuietBelowTheReclaimableThreshold(t *testing.T) {
+	t.Parallel()
+
+	v := classifyDrift(sweepWithReclaimable(reclaimableAlertThreshold-1, 0))
+
+	assert.Empty(t, v.Op, "a handful of reclaimable items is the expected residue of an interrupted deploy")
+}
+
+func TestClassifyDrift_AlertsOnceReclaimableDriftAccumulates(t *testing.T) {
+	t.Parallel()
+
+	v := classifyDrift(sweepWithReclaimable(reclaimableAlertThreshold, 0))
+
+	require.Equal(t, opDriftReclaimable, v.Op,
+		"abandoned deploy sessions accrued for 33 days with nothing watching; a report-only cron that never "+
+			"raises on them is indistinguishable from not looking")
+	assert.False(t, v.Fails, "reclaimable drift is storage cost, not an outage: alert, do not fail the run")
+	require.Error(t, v.Err)
+	assert.Contains(t, v.Err.Error(), "artemis reconcile")
+	assert.Contains(t, v.Err.Error(), "www.freecode.camp",
+		"a fleet-wide count with a literal <site> placeholder tells the operator nothing about where to look")
+}
+
+func TestClassifyDrift_AliasedMissingOutranksTheReclaimableThreshold(t *testing.T) {
+	t.Parallel()
+
+	res := sweepWithReclaimable(reclaimableAlertThreshold+50, 0)
+	res.Reports[0].Aliased = []string{"d1"}
+
+	v := classifyDrift(res)
+
+	assert.Equal(t, opDriftAliasedMissing, v.Op,
+		"a live site serving nothing must not be masked by a large but harmless reclaimable count")
+}
+
+func TestEveryDriftVerdictOpIsCronShaped(t *testing.T) {
+	t.Parallel()
+
+	for _, op := range []string{opDriftSweep, opDriftSelfCheck, opDriftUnreadable, opDriftAliasedMissing, opDriftReclaimable} {
+		assert.True(t, observability.IsCronShaped(op),
+			"op %s bypasses the transient-rate tracker only if cronShapedOps lists it; this test lives "+
+				"beside the op constants so adding a sixth verdict here fails until the map learns it — "+
+				"the observability-side test could only restate the map against itself", op)
+	}
+}

cmd/artemis/driftfixtures_test.go

@@ -2,71 +2,8 @@ package main
 
 import (
 	"context"
-	"errors"
-	"log/slog"
-	"sync"
-	"time"
-
-	"github.com/freeCodeCamp/artemis/internal/gc"
 )
 
-type msgCapture struct {
-	mu   sync.Mutex
-	msgs []string
-}
-
-func (h *msgCapture) Enabled(context.Context, slog.Level) bool { return true }
-func (h *msgCapture) Handle(_ context.Context, r slog.Record) error {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	h.msgs = append(h.msgs, r.Message)
-	return nil
-}
-func (h *msgCapture) WithAttrs([]slog.Attr) slog.Handler { return h }
-func (h *msgCapture) WithGroup(string) slog.Handler      { return h }
-
-func (h *msgCapture) saw(msg string) bool {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	for _, m := range h.msgs {
-		if m == msg {
-			return true
-		}
-	}
-	return false
-}
-
-type driftingStore struct{ nopReconcileStore }
-
-func (driftingStore) DeploysForSite(context.Context, string) ([]gc.Deploy, error) {
-	return []gc.Deploy{{ID: "ghost", Mtime: time.Now().Add(-30 * 24 * time.Hour)}}, nil
-}
-
-func (driftingStore) AliasTargets(context.Context, string) (map[string]struct{}, time.Time, error) {
-	return map[string]struct{}{"ghost": {}}, time.Time{}, nil
-}
-
-func (driftingStore) RecordTombstone(context.Context, string, string, int64) error {
-	return errors.New("pg down")
-}
-
 type staticLister struct{ keys []string }
 
 func (l staticLister) ListPrefix(context.Context, string) ([]string, error) { return l.keys, nil }
-
-type passthroughSession struct{}
-
-func (passthroughSession) WithSiteLock(_ context.Context, _ string, fn func() error) error {
-	return fn()
-}
-func (passthroughSession) Close(context.Context) {}
-
-type passthroughLocker struct{}
-
-func (passthroughLocker) NewLockSession(context.Context) (gc.LockSession, error) {
-	return passthroughSession{}, nil
-}
-
-type nopMover struct{}
-
-func (nopMover) MovePrefix(context.Context, string, string) (int, error) { return 0, nil }

cmd/artemis/gcwire.go

@@ -23,7 +23,6 @@ type auditRecorder interface {
 var captureAuditFailure = observability.CaptureBackground
 
 var (
-	_ handler.SiteChangeEmitter = (*pg.Repo)(nil)
 	_ handler.TombstoneStore    = (*pg.Repo)(nil)
 	_ handler.TrashStore        = (*pg.Repo)(nil)
 	_ handler.DeployIndexWriter = (*pg.Repo)(nil)
@@ -39,10 +38,10 @@ func wirePGRepo(h *handler.Handlers, repo *pg.Repo) {
 	if repo == nil {
 		return
 	}
-	h.Outbox = repo
 	h.Tombstones = repo
 	h.Trash = repo
 	h.Index = repo
+	h.Pending = repo
 	h.Locker = repo
 	h.Audit = repo
 }
@@ -130,18 +129,56 @@ type aliasGetter interface {
 	GetAlias(ctx context.Context, aliasKey string) (string, error)
 }
 
-func newLiveAliasReader(getter aliasGetter, formats ...string) (func(context.Context, string) (map[string]struct{}, error), error) {
-	fmts := make([]string, 0, len(formats))
+func siteSegment(format string) (string, error) {
+	slash := strings.IndexByte(format, '/')
+	if slash < 0 {
+		return "", fmt.Errorf("key format %q must contain '/' after the site segment", format)
+	}
+	return format[:slash], nil
+}
+
+func aliasTails(deployFormat string, formats ...string) ([]string, error) {
+	deploySeg, err := siteSegment(deployFormat)
+	if err != nil {
+		return nil, fmt.Errorf("DEPLOY_PREFIX_FORMAT: %w", err)
+	}
+	tails := make([]string, 0, len(formats))
 	for _, f := range formats {
 		if !strings.Contains(f, "<site>") {
 			return nil, fmt.Errorf("alias key format %q must contain <site>", f)
 		}
-		fmts = append(fmts, f)
+		seg, err := siteSegment(f)
+		if err != nil {
+			return nil, err
+		}
+		if seg != deploySeg {
+			return nil, fmt.Errorf(
+				"alias key format %q has site segment %q but DEPLOY_PREFIX_FORMAT %q has %q: "+
+					"the GC sweep enumerates storage dirnames rendered from the deploy prefix, so an alias "+
+					"key under a different site segment is unreachable and would 404 for every site",
+				f, seg, deployFormat, deploySeg)
+		}
+		tail := f[len(seg)+1:]
+		if strings.Contains(tail, "<site>") {
+			return nil, fmt.Errorf(
+				"alias key format %q keeps a <site> token after its site segment: only the segment is "+
+					"rendered from the dirname, so the rest is fetched literally and 404s for every site",
+				f)
+		}
+		tails = append(tails, tail)
+	}
+	return tails, nil
+}
+
+func newLiveAliasReader(getter aliasGetter, deployFormat string, formats ...string) (func(context.Context, string) (map[string]struct{}, error), error) {
+	tails, err := aliasTails(deployFormat, formats...)
+	if err != nil {
+		return nil, err
 	}
-	return func(ctx context.Context, site string) (map[string]struct{}, error) {
+	return func(ctx context.Context, dirname string) (map[string]struct{}, error) {
 		out := map[string]struct{}{}
-		for _, f := range fmts {
-			v, err := getter.GetAlias(ctx, strings.ReplaceAll(f, "<site>", site))
+		for _, tail := range tails {
+			v, err := getter.GetAlias(ctx, dirname+"/"+tail)
 			if err != nil {
 				if r2.IsNotFound(err) {
 					continue
@@ -177,7 +214,8 @@ func newGCWiring(cfg *config.Config, repo *pg.Repo, r2c *r2.Client) (*gcWiring,
 	if err != nil {
 		return nil, err
 	}
-	liveAliases, err := newLiveAliasReader(r2c, cfg.Aliases.ProductionKeyFormat, cfg.Aliases.PreviewKeyFormat)
+	liveAliases, err := newLiveAliasReader(r2c, cfg.DeployPrefixFormat,
+		cfg.Aliases.ProductionKeyFormat, cfg.Aliases.PreviewKeyFormat)
 	if err != nil {
 		return nil, err
 	}
@@ -188,6 +226,7 @@ func newGCWiring(cfg *config.Config, repo *pg.Repo, r2c *r2.Client) (*gcWiring,
 			Mover:        r2c,
 			Locker:       repo,
 			LiveAliases:  liveAliases,
+			Pending:      repo,
 			Policy:       gcPolicy(cfg.Cleanup),
 			BlastCap:     cfg.Cleanup.BlastCap,
 			DeployPrefix: layout.deployPrefix,
@@ -215,6 +254,7 @@ func newGCWiring(cfg *config.Config, repo *pg.Repo, r2c *r2.Client) (*gcWiring,
 			Deleter:   r2c,
 			Recovery:  time.Duration(cfg.Cleanup.RecoveryDays) * 24 * time.Hour,
 			TrashBase: cfg.Cleanup.TrashPrefix,
+			BlastCap:  cfg.Cleanup.BlastCap,
 			Now:       time.Now,
 			Locker:    repo,
 			Audit:     gcPurgeAuditor{repo: repo},