feat: durable execution model

.gitignore

@@ -1,6 +1,7 @@
 # Binaries
 /bin/
 /artemis
+/loadgen
 *.exe
 *.test
 *.out

cmd/artemis/gcwire.go

@@ -0,0 +1,122 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/freeCodeCamp/artemis/internal/backfill"
+	"github.com/freeCodeCamp/artemis/internal/config"
+	"github.com/freeCodeCamp/artemis/internal/gc"
+	"github.com/freeCodeCamp/artemis/internal/handler"
+	"github.com/freeCodeCamp/artemis/internal/pg"
+	"github.com/freeCodeCamp/artemis/internal/r2"
+	"github.com/freeCodeCamp/artemis/internal/registry/valkey"
+)
+
+var (
+	_ handler.SiteChangeEmitter = (*pg.Repo)(nil)
+	_ handler.TombstoneStore    = (*pg.Repo)(nil)
+	_ handler.RepoStore         = (*pg.RepoQueue)(nil)
+	_ backfill.Lister           = (*r2.Client)(nil)
+	_ backfill.Indexer          = (*pg.Repo)(nil)
+	_ pg.SitesSource            = (*valkey.Store)(nil)
+)
+
+func openRepoQueue(pgDB *pg.DB) (handler.RepoStore, error) {
+	if pgDB == nil {
+		return nil, fmt.Errorf("repo-creation feature requires DATABASE_URL")
+	}
+	return pg.NewRepoQueue(pgDB), nil
+}
+
+const deployIDToken = "<ts>-<sha>"
+
+type gcLayout struct {
+	sitePrefix   func(site string) string
+	deployPrefix func(site, id string) string
+	trashPrefix  func(site, id string) string
+}
+
+func newGCLayout(format, trashBase string) (gcLayout, error) {
+	idx := strings.Index(format, deployIDToken)
+	if idx < 0 {
+		return gcLayout{}, fmt.Errorf("DEPLOY_PREFIX_FORMAT %q must contain %s", format, deployIDToken)
+	}
+	head := format[:idx]
+	tail := format[idx+len(deployIDToken):]
+	slash := strings.IndexByte(head, '/')
+	if slash < 0 {
+		return gcLayout{}, fmt.Errorf("DEPLOY_PREFIX_FORMAT %q must contain '/' after the site segment", format)
+	}
+	subPath := head[slash+1:]
+	if trashBase == "" {
+		trashBase = "_trash/"
+	}
+	return gcLayout{
+		sitePrefix: func(site string) string { return site + "/" + subPath },
+		deployPrefix: func(site, id string) string {
+			p := site + "/" + subPath + id + tail
+			if !strings.HasSuffix(p, "/") {
+				p += "/"
+			}
+			return p
+		},
+		trashPrefix: func(site, id string) string { return trashBase + site + "/" + id + "/" },
+	}, nil
+}
+
+func gcPolicy(c config.CleanupConfig) gc.Policy {
+	return gc.Policy{
+		RecentKeep:    c.RecentKeep,
+		Grace:         c.Grace,
+		Retention:     time.Duration(c.RetentionDays) * 24 * time.Hour,
+		ServeCacheTTL: c.ServeCacheTTL,
+	}
+}
+
+type gcWiring struct {
+	Repo       *pg.Repo
+	SiteGC     *gc.SiteGC
+	Reconciler *gc.Reconciler
+	Purge      *gc.TombstonePurge
+}
+
+func newGCWiring(cfg *config.Config, repo *pg.Repo, r2c *r2.Client, metrics *gc.Metrics) (*gcWiring, error) {
+	layout, err := newGCLayout(cfg.DeployPrefixFormat, cfg.Cleanup.TrashPrefix)
+	if err != nil {
+		return nil, err
+	}
+	return &gcWiring{
+		Repo: repo,
+		SiteGC: &gc.SiteGC{
+			Store:        repo,
+			Mover:        r2c,
+			Policy:       gcPolicy(cfg.Cleanup),
+			BlastCap:     cfg.Cleanup.BlastCap,
+			DeployPrefix: layout.deployPrefix,
+			TrashPrefix:  layout.trashPrefix,
+			Now:          time.Now,
+			Metrics:      metrics,
+		},
+		Reconciler: &gc.Reconciler{
+			Lister:       r2c,
+			Store:        repo,
+			Mover:        r2c,
+			Grace:        cfg.Cleanup.Grace,
+			SitePrefix:   layout.sitePrefix,
+			DeployPrefix: layout.deployPrefix,
+			TrashPrefix:  layout.trashPrefix,
+			Now:          time.Now,
+			Metrics:      metrics,
+		},
+		Purge: &gc.TombstonePurge{
+			Store:     repo,
+			Deleter:   r2c,
+			Recovery:  time.Duration(cfg.Cleanup.RecoveryDays) * 24 * time.Hour,
+			TrashBase: cfg.Cleanup.TrashPrefix,
+			Now:       time.Now,
+			Metrics:   metrics,
+		},
+	}, nil
+}

cmd/artemis/gcwire_test.go

@@ -0,0 +1,130 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/freeCodeCamp/artemis/internal/config"
+	"github.com/freeCodeCamp/artemis/internal/pg"
+	"github.com/freeCodeCamp/artemis/internal/r2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOpenRepoQueue_RequiresDatabase(t *testing.T) {
+	q, err := openRepoQueue(nil)
+	require.Error(t, err, "repo feature without a database must be rejected at boot")
+	require.Nil(t, q)
+}
+
+func TestOpenRepoQueue_IsPostgresBacked(t *testing.T) {
+	q, err := openRepoQueue(&pg.DB{})
+	require.NoError(t, err)
+	_, ok := q.(*pg.RepoQueue)
+	assert.True(t, ok, "repo queue must be backed by pg.RepoQueue")
+}
+
+func TestBootWiringProdLayout(t *testing.T) {
+	cases := []struct {
+		name       string
+		format     string
+		trashBase  string
+		site       string
+		id         string
+		wantSite   string
+		wantDeploy string
+		wantTrash  string
+	}{
+		{
+			name:       "default-dev-layout",
+			format:     "<site>/deploys/<ts>-<sha>/",
+			trashBase:  "_trash/",
+			site:       "www",
+			id:         "20260101-000000-abc1234",
+			wantSite:   "www/deploys/",
+			wantDeploy: "www/deploys/20260101-000000-abc1234/",
+			wantTrash:  "_trash/www/20260101-000000-abc1234/",
+		},
+		{
+			name:       "prod-dirname-layout",
+			format:     "<site>.freecode.camp/deploys/<ts>-<sha>/",
+			trashBase:  "_trash/",
+			site:       "www.freecode.camp",
+			id:         "20260101-000000-abc1234",
+			wantSite:   "www.freecode.camp/deploys/",
+			wantDeploy: "www.freecode.camp/deploys/20260101-000000-abc1234/",
+			wantTrash:  "_trash/www.freecode.camp/20260101-000000-abc1234/",
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			l, err := newGCLayout(tc.format, tc.trashBase)
+			require.NoError(t, err)
+			assert.Equal(t, tc.wantSite, l.sitePrefix(tc.site), "sitePrefix")
+			assert.Equal(t, tc.wantDeploy, l.deployPrefix(tc.site, tc.id), "deployPrefix")
+			assert.Equal(t, tc.wantTrash, l.trashPrefix(tc.site, tc.id), "trashPrefix")
+		})
+	}
+}
+
+func TestBootWiring_LayoutRejectsBadFormat(t *testing.T) {
+	_, err := newGCLayout("<site>/deploys/", "_trash/")
+	require.Error(t, err, "format without the deploy-id token must be rejected")
+}
+
+func TestNewGCWiring_PlumbsBlastCapAndPrefixes(t *testing.T) {
+	cfg := &config.Config{
+		DeployPrefixFormat: "<site>/deploys/<ts>-<sha>/",
+		Cleanup: config.CleanupConfig{
+			BlastCap:      5,
+			RetentionDays: 7,
+			RecoveryDays:  3,
+			TrashPrefix:   "_trash/",
+		},
+	}
+	repo := &pg.Repo{}
+	r2c := &r2.Client{}
+
+	w, err := newGCWiring(cfg, repo, r2c, nil)
+	require.NoError(t, err)
+	require.NotNil(t, w)
+
+	assert.Same(t, repo, w.Repo, "repo must be plumbed through")
+	assert.Equal(t, 5, w.SiteGC.BlastCap, "BlastCap=0 would disable the mass-delete safety cap")
+	assert.Equal(t, 7*24*time.Hour, w.SiteGC.Policy.Retention, "policy retention must derive from RetentionDays")
+	assert.Equal(t, "_trash/", w.Purge.TrashBase, "purge must scan the configured trash base")
+	assert.Equal(t, 3*24*time.Hour, w.Purge.Recovery, "recovery window must derive from RecoveryDays")
+
+	require.NotNil(t, w.SiteGC.DeployPrefix)
+	require.NotNil(t, w.SiteGC.TrashPrefix)
+	require.NotNil(t, w.Reconciler.SitePrefix)
+	require.NotNil(t, w.Reconciler.DeployPrefix)
+
+	assert.Equal(t, "www/deploys/id/", w.SiteGC.DeployPrefix("www", "id"),
+		"a wrong deploy-prefix closure would mass-move the wrong R2 prefix")
+	assert.Equal(t, "_trash/www/id/", w.SiteGC.TrashPrefix("www", "id"))
+	assert.Equal(t, "www/deploys/", w.Reconciler.SitePrefix("www"))
+}
+
+func TestNewGCWiring_RejectsBadFormat(t *testing.T) {
+	cfg := &config.Config{
+		DeployPrefixFormat: "<site>/deploys/",
+		Cleanup:            config.CleanupConfig{BlastCap: 5, TrashPrefix: "_trash/"},
+	}
+	w, err := newGCWiring(cfg, &pg.Repo{}, &r2.Client{}, nil)
+	require.Error(t, err, "a format missing the deploy-id token must fail boot wiring, not produce a degenerate prefix fn")
+	require.Nil(t, w)
+}
+
+func TestGCPolicyFromConfig(t *testing.T) {
+	p := gcPolicy(config.CleanupConfig{
+		RecentKeep:    3,
+		Grace:         time.Hour,
+		RetentionDays: 7,
+		ServeCacheTTL: 15 * time.Second,
+	})
+	assert.Equal(t, 3, p.RecentKeep)
+	assert.Equal(t, time.Hour, p.Grace)
+	assert.Equal(t, 7*24*time.Hour, p.Retention)
+	assert.Equal(t, 15*time.Second, p.ServeCacheTTL)
+}

cmd/artemis/gcworkflows.go

@@ -0,0 +1,113 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/freeCodeCamp/artemis/internal/observability"
+	"github.com/freeCodeCamp/artemis/internal/pg"
+	"github.com/freeCodeCamp/artemis/internal/worker"
+)
+
+const (
+	topicSiteReconcile = "site.reconcile"
+	cronTombstonePurge = "0 3 * * *"
+	relayInterval      = 5 * time.Second
+)
+
+func runRelayLoop(ctx context.Context, relay *worker.Relay, interval time.Duration, metrics *worker.Metrics) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			n, err := relay.RunOnce(ctx)
+			metrics.ObserveRelay(n, err)
+			if err != nil {
+				slog.Error("relay.run", "err", err)
+				observability.CaptureBackground("relay.run", err)
+			}
+		}
+	}
+}
+
+func observeWorkflow(metrics *worker.Metrics, name string, fn worker.Handler) worker.Handler {
+	return func(ctx context.Context, input map[string]any) error {
+		err := fn(ctx, input)
+		outcome := "ok"
+		if err != nil {
+			outcome = "failed"
+		}
+		metrics.ObserveRun(name, outcome)
+		return err
+	}
+}
+
+func gcWorkflowDefs(gcw *gcWiring, dryRun bool, metrics *worker.Metrics) []worker.WorkflowDef {
+	return []worker.WorkflowDef{
+		{
+			Name:           worker.WorkflowGCSite,
+			ConcurrencyKey: worker.ConcurrencyKeySite,
+			EventTriggers:  []string{pg.TopicSiteChanged},
+			Handler: observeWorkflow(metrics, worker.WorkflowGCSite, func(ctx context.Context, input map[string]any) error {
+				site, err := siteFromInput(input)
+				if err != nil {
+					return err
+				}
+				if _, err := gcw.SiteGC.Run(ctx, site, dryRun); err != nil {
+					observability.CaptureBackground("gc.site.run", err)
+					return err
+				}
+				return nil
+			}),
+		},
+		{
+			Name: worker.WorkflowTombstonePurge,
+			Cron: []string{cronTombstonePurge},
+			Handler: observeWorkflow(metrics, worker.WorkflowTombstonePurge, func(ctx context.Context, _ map[string]any) error {
+				if _, err := gcw.Purge.Run(ctx, dryRun); err != nil {
+					observability.CaptureBackground("tombstone.purge", err)
+					return err
+				}
+				return nil
+			}),
+		},
+		{
+			Name:           worker.WorkflowReconcile,
+			ConcurrencyKey: worker.ConcurrencyKeySite,
+			EventTriggers:  []string{topicSiteReconcile},
+			Handler: observeWorkflow(metrics, worker.WorkflowReconcile, func(ctx context.Context, input map[string]any) error {
+				site, err := siteFromInput(input)
+				if err != nil {
+					return err
+				}
+				if _, err := gcw.Reconciler.ReconcileSite(ctx, site); err != nil {
+					observability.CaptureBackground("reconcile.run", err)
+					return err
+				}
+				return nil
+			}),
+		},
+	}
+}
+
+func siteFromInput(input map[string]any) (string, error) {
+	s, ok := input["site"].(string)
+	if !ok || s == "" {
+		return "", errors.New("workflow input missing site")
+	}
+	return s, nil
+}
+
+func registerGCWorkflows(rt *worker.Runtime, gcw *gcWiring, dryRun bool, metrics *worker.Metrics) error {
+	for _, def := range gcWorkflowDefs(gcw, dryRun, metrics) {
+		if err := rt.Register(def); err != nil {
+			return err
+		}
+	}
+	return nil
+}