amadan

carlos / platform Public

stopevidence: composite stop evidence and cleanup-only entry point (claim Plan 3 Task 6)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 8e90a399d644accbe8f79baf53ebca22b58a5548 parent b3c19f9
2 files changed, +298 −0
  • internal/stopevidence/evidence.go +128 −0
  • internal/stopevidence/evidence_linux_test.go +170 −0
diff --git a/internal/stopevidence/evidence.go b/internal/stopevidence/evidence.go
new file mode 100644
index 000000000..4c26a13c8
--- /dev/null
+++ b/internal/stopevidence/evidence.go
@@ -0,0 +1,128 @@
+package stopevidence
+
+import (
+ "context"
+ "errors"
+ "path/filepath"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/controlname"
+)
+
+var (
+ ErrFence = errors.New("stopevidence: matching stop fence required")
+ ErrNoCgroupEvidence = errors.New("stopevidence: cgroup observation required")
+ ErrPopulated = errors.New("stopevidence: cgroup populated")
+ ErrMembers = errors.New("stopevidence: cgroup has members")
+ ErrReplicatorsOpen = errors.New("stopevidence: replicators not closed")
+ ErrFDHolders = errors.New("stopevidence: database descriptors still held")
+)
+
+type Inputs struct {
+ Host string
+ Generation uint64
+ Fence *activator.StopEvidence
+ Cgroup interface {
+ Populated() (bool, error)
+ Members() ([]int, error)
+ }
+ DBPaths []string
+ ProcRoot string
+ ReplicatorsClosed bool
+}
+
+type Evidence struct {
+ Host string
+ Generation uint64
+ CheckedAt time.Time
+ FenceClosed bool
+ SpawnAttemptsDrained bool
+ CgroupUnpopulated bool
+ CgroupMembersEmpty bool
+ ReplicatorsClosed bool
+ FDScanComplete bool
+ CheckedPaths []string
+ ProcRoot string
+}
+
+func Collect(ctx context.Context, in Inputs) (Evidence, error) {
+ if ctx == nil {
+ return Evidence{}, ErrInvalidInput
+ }
+ if err := ctx.Err(); err != nil {
+ return Evidence{}, err
+ }
+ if !controlname.ValidHostShape(in.Host) || in.Generation == 0 {
+ return Evidence{}, ErrInvalidInput
+ }
+ if in.Fence == nil || !in.Fence.Matches(in.Host, in.Generation) {
+ return Evidence{}, ErrFence
+ }
+ if in.Cgroup == nil {
+ return Evidence{}, ErrNoCgroupEvidence
+ }
+ if !in.ReplicatorsClosed {
+ return Evidence{}, ErrReplicatorsOpen
+ }
+ if len(in.DBPaths) == 0 {
+ return Evidence{}, ErrInvalidInput
+ }
+ seen := map[string]bool{}
+ for _, db := range in.DBPaths {
+ if !filepath.IsAbs(db) {
+ return Evidence{}, ErrInvalidInput
+ }
+ db = filepath.Clean(db)
+ seen[db], seen[db+"-wal"], seen[db+"-shm"] = true, true, true
+ }
+ paths := make([]string, 0, len(seen))
+ for path := range seen {
+ paths = append(paths, path)
+ }
+ sort.Strings(paths)
+ populated, err := in.Cgroup.Populated()
+ if err != nil {
+ return Evidence{}, err
+ }
+ if populated {
+ return Evidence{}, ErrPopulated
+ }
+ members, err := in.Cgroup.Members()
+ if err != nil {
+ return Evidence{}, err
+ }
+ if len(members) != 0 {
+ return Evidence{}, ErrMembers
+ }
+ root := in.ProcRoot
+ if root == "" {
+ root = "/proc"
+ }
+ holders, err := scanFDs(ctx, root, paths)
+ if err != nil {
+ return Evidence{}, err
+ }
+ if len(holders) != 0 {
+ return Evidence{}, ErrFDHolders
+ }
+ if err := ctx.Err(); err != nil {
+ return Evidence{}, err
+ }
+ return Evidence{
+ Host: in.Host, Generation: in.Generation, CheckedAt: time.Now().UTC(),
+ FenceClosed: true, SpawnAttemptsDrained: true,
+ CgroupUnpopulated: true, CgroupMembersEmpty: true,
+ ReplicatorsClosed: true, FDScanComplete: true,
+ CheckedPaths: paths, ProcRoot: root,
+ }, nil
+}
+
+func Cleanup(ctx context.Context, c *cgroup.Cleanup) error {
+ if c == nil {
+ return ErrInvalidInput
+ }
+ return c.KillAndEmpty(ctx)
+}
diff --git a/internal/stopevidence/evidence_linux_test.go b/internal/stopevidence/evidence_linux_test.go
new file mode 100644
index 000000000..c80347c28
--- /dev/null
+++ b/internal/stopevidence/evidence_linux_test.go
@@ -0,0 +1,170 @@
+//go:build linux
+
+package stopevidence
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "reflect"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+type evidenceProc struct{ exited atomic.Bool }
+
+func (p *evidenceProc) Stop(context.Context) error { p.exited.Store(true); return nil }
+func (p *evidenceProc) Exited() bool { return p.exited.Load() }
+
+type evidenceRep struct{}
+
+func (evidenceRep) Err() error { return nil }
+func (evidenceRep) Close(context.Context) error { return nil }
+
+type evidenceLease struct{}
+
+func (evidenceLease) Release(context.Context) error { return nil }
+func (evidenceLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) {
+ return evidenceRep{}, nil
+}
+
+type evidenceBackend struct{}
+
+func (evidenceBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) {
+ return evidenceLease{}, nil
+}
+func (evidenceBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
+func (evidenceBackend) Seed(context.Context, registry.Route) error { return nil }
+func (evidenceBackend) Start(context.Context, registry.Route) (activator.Proc, string, error) {
+ return &evidenceProc{}, "test-build", nil
+}
+func (evidenceBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
+func (evidenceBackend) HasSidecar(registry.Route) bool { return false }
+func (evidenceBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
+ return nil, errors.New("unexpected sidecar")
+}
+func (evidenceBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) {
+ return nil, "", errors.New("unexpected worker")
+}
+
+func fenceFixture(t *testing.T, host string, generation uint64) *activator.StopEvidence {
+ t.Helper()
+ dir := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ a := activator.New(activator.Config{Bucket: "test", Idle: time.Hour}, evidenceBackend{}, reg)
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ a.Shutdown(ctx)
+ })
+ rt := registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: filepath.Join(dir, "app.sock"), DBPath: filepath.Join(dir, "data.db")}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+ defer cancel()
+ called := false
+ w := httptest.NewRecorder()
+ r := httptest.NewRequest(http.MethodGet, "http://"+host+"/", nil).WithContext(ctx)
+ if !a.ServeHTTP(w, r, rt, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { called = true; w.WriteHeader(http.StatusNoContent) })) || !called {
+ t.Fatalf("wake failed: %d %s", w.Code, w.Body.String())
+ }
+ proof, err := a.FenceAndStop(ctx, host, generation)
+ if err != nil || !proof.Matches(host, generation) {
+ t.Fatal("fence proof", err)
+ }
+ return proof
+}
+
+type evidenceGroup struct {
+ populated bool
+ members []int
+ populationErr, membersErr error
+}
+
+func (g *evidenceGroup) Populated() (bool, error) { return g.populated, g.populationErr }
+func (g *evidenceGroup) Members() ([]int, error) { return g.members, g.membersErr }
+
+func TestCollectRequiresEveryCondition(t *testing.T) {
+ host := "shop.example"
+ proof := fenceFixture(t, host, 7)
+ failure := errors.New("read failed")
+ for _, tc := range []struct {
+ name string
+ change func(*testing.T, *Inputs, *evidenceGroup)
+ want error
+ }{
+ {"success", func(t *testing.T, in *Inputs, g *evidenceGroup) {}, nil},
+ {"no-fence", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.Fence = nil }, ErrFence},
+ {"empty-fence", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.Fence = &activator.StopEvidence{} }, ErrFence},
+ {"wrong-host", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.Host = "other.example" }, ErrFence},
+ {"wrong-generation", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.Generation = 8 }, ErrFence},
+ {"no-generation", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.Generation = 0 }, ErrInvalidInput},
+ {"no-group", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.Cgroup = nil }, ErrNoCgroupEvidence},
+ {"population", func(t *testing.T, in *Inputs, g *evidenceGroup) { g.populated = true }, ErrPopulated},
+ {"members", func(t *testing.T, in *Inputs, g *evidenceGroup) { g.members = []int{55} }, ErrMembers},
+ {"population-error", func(t *testing.T, in *Inputs, g *evidenceGroup) { g.populationErr = failure }, failure},
+ {"members-error", func(t *testing.T, in *Inputs, g *evidenceGroup) { g.membersErr = failure }, failure},
+ {"replicator-open", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.ReplicatorsClosed = false }, ErrReplicatorsOpen},
+ {"no-paths", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.DBPaths = nil }, ErrInvalidInput},
+ {"relative-path", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.DBPaths = []string{"data.db"} }, ErrInvalidInput},
+ {"proc-read-error", func(t *testing.T, in *Inputs, g *evidenceGroup) { in.ProcRoot = filepath.Join(in.ProcRoot, "missing") }, ErrIncomplete},
+ {"db-holder", func(t *testing.T, in *Inputs, g *evidenceGroup) { addHolder(t, in, "") }, ErrFDHolders},
+ {"wal-holder", func(t *testing.T, in *Inputs, g *evidenceGroup) { addHolder(t, in, "-wal") }, ErrFDHolders},
+ {"shm-holder", func(t *testing.T, in *Inputs, g *evidenceGroup) { addHolder(t, in, "-shm") }, ErrFDHolders},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ g := &evidenceGroup{}
+ in := Inputs{Host: host, Generation: 7, Fence: proof, Cgroup: g, DBPaths: []string{filepath.Join(t.TempDir(), "data.db")}, ProcRoot: t.TempDir(), ReplicatorsClosed: true}
+ tc.change(t, &in, g)
+ e, err := Collect(t.Context(), in)
+ if tc.want != nil {
+ if !errors.Is(err, tc.want) || !reflect.DeepEqual(e, Evidence{}) {
+ t.Fatalf("evidence on failure: %+v %v", e, err)
+ }
+ return
+ }
+ if err != nil || e.Host != host || e.Generation != 7 || !e.FenceClosed || !e.SpawnAttemptsDrained || !e.CgroupUnpopulated || !e.CgroupMembersEmpty || !e.ReplicatorsClosed || !e.FDScanComplete || e.CheckedAt.IsZero() || len(e.CheckedPaths) != 3 {
+ t.Fatalf("incomplete evidence: %+v %v", e, err)
+ }
+ g.populated = true
+ if _, err := Collect(t.Context(), in); !errors.Is(err, ErrPopulated) {
+ t.Fatal("cached evidence", err)
+ }
+ })
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ if e, err := Collect(ctx, Inputs{}); !errors.Is(err, context.Canceled) || !reflect.DeepEqual(e, Evidence{}) {
+ t.Fatal(e, err)
+ }
+ if err := Cleanup(t.Context(), nil); !errors.Is(err, ErrInvalidInput) {
+ t.Fatal(err)
+ }
+ if err := Cleanup(t.Context(), &cgroup.Cleanup{}); !errors.Is(err, cgroup.ErrStrictState) {
+ t.Fatal(err)
+ }
+}
+
+func addHolder(t *testing.T, in *Inputs, suffix string) {
+ t.Helper()
+ dir := filepath.Join(in.ProcRoot, "55", "fd")
+ if err := os.MkdirAll(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(in.DBPaths[0]+suffix, filepath.Join(dir, "4")); err != nil {
+ t.Fatal(err)
+ }
+}