amadan

carlos / platform Public

Merge origin/main into feat/alerts-sns

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude afafc5145c0ecf63532e20e078bce9031f16b50c parent fb8f919parent 3ca43d3
17 files changed, +3014 −5
  • AGENTS.md +6 −1
  • infra/modules/region-host/main.tf +14 −0
  • infra/modules/region-remote-host/main.tf +14 −0
  • internal/activator/activator.go +119 −0
  • internal/activator/converge_adopted_test.go +203 −0
  • internal/edge/static.go +301 −2
  • internal/edge/static_cache_test.go +596 −0
  • internal/host/agent.go +111 −2
  • internal/host/converge_adopted_test.go +416 −0
  • internal/lease/export_test.go +9 −0
  • internal/lease/leasetest/bucket.go +340 −0
  • internal/lease/leasetest/bucket_test.go +15 −0
  • internal/lease/leasetest/contract.go +245 −0
  • internal/lease/leasetest/s3server.go +222 −0
  • internal/lease/leasetest/s3server_test.go +38 −0
  • internal/lease/versioned.go +268 −0
  • internal/lease/versioned_test.go +97 −0

CI passed — run details

diff --git a/AGENTS.md b/AGENTS.md
index cdca5b6a0..455dd458c 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -344,7 +344,12 @@ anything.
poll it at 2s and run the full pass set. Direct-bucket promotes (pinfra
scripts) don't bump but canary/* channels are name-scoped fast anyway.
Blocked stages relay to a root pass (`carlos-stage.path` — must be armed;
- update-platform-binary.sh does it). `carlos deploy` is the member one-command
+ update-platform-binary.sh does it). That root pass stages, repoints and
+ stamps in ~8s but cannot cycle the agent's exec children, so the agent's
+ 2s tick finishes the job: a warm instance whose route is stamped with the
+ version its symlink names is cycled at once (`convergeAdopted`, #231). The
+ activator sweep's 90s grace is only the backstop. `carlos deploy` is the
+ member one-command
form: ship → promote → wait until `X-Carlos-Version` on the URL reports the
shipped build over 2xx. Verify deploys with that header (`curl -sI
--resolve`), never a cached answer. `-I` is fine for the VERSION header, but
diff --git a/infra/modules/region-host/main.tf b/infra/modules/region-host/main.tf
index 14b7724a6..97a13e100 100644
--- a/infra/modules/region-host/main.tf
+++ b/infra/modules/region-host/main.tf
@@ -174,6 +174,20 @@ resource "aws_iam_role_policy" "host_buckets" {
Action = ["s3:ListBucket", "s3:GetBucketLocation"]
Resource = [var.deployment_bucket_arn, var.data_bucket_arn]
},
+ # Host lifecycle claim, Plan 2 (spec C3, §6.2, §6.5): exact-version
+ # list, read and delete on this box's OWN data bucket. Sibling-region
+ # moves/* grants are unchanged: the destination region retires its
+ # own staging (spec N13).
+ {
+ Effect = "Allow"
+ Action = ["s3:GetObjectVersion", "s3:DeleteObjectVersion"]
+ Resource = ["${var.data_bucket_arn}/*"]
+ },
+ {
+ Effect = "Allow"
+ Action = ["s3:ListBucketVersions"]
+ Resource = [var.data_bucket_arn]
+ },
{
Effect = "Allow"
Action = ["ssm:GetParameter"]
diff --git a/infra/modules/region-remote-host/main.tf b/infra/modules/region-remote-host/main.tf
index d18648e76..d86783977 100644
--- a/infra/modules/region-remote-host/main.tf
+++ b/infra/modules/region-remote-host/main.tf
@@ -91,6 +91,20 @@ resource "aws_iam_user_policy" "host" {
Action = ["s3:ListBucket", "s3:GetBucketLocation"]
Resource = [aws_s3_bucket.data.arn]
},
+ # Host lifecycle claim, Plan 2 (spec C3, §6.2, §6.5): exact-version
+ # list, read and delete on this box's OWN data bucket. Sibling-region
+ # moves/* grants are unchanged: the destination region retires its
+ # own staging (spec N13).
+ {
+ Effect = "Allow"
+ Action = ["s3:GetObjectVersion", "s3:DeleteObjectVersion"]
+ Resource = ["${aws_s3_bucket.data.arn}/*"]
+ },
+ {
+ Effect = "Allow"
+ Action = ["s3:ListBucketVersions"]
+ Resource = [aws_s3_bucket.data.arn]
+ },
],
[for arn in var.sibling_staging_arns : {
Effect = "Allow"
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 16ef6357a..15e9b11f5 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1381,6 +1381,12 @@ func (a *Service) Bounce(host string) bool {
if e == nil || e.state != actAwake {
return false
}
+ return a.bounceLocked(e)
+}
+
+// bounceLocked is Bounce's stop for an entry the caller has found awake,
+// under a.mu.
+func (a *Service) bounceLocked(e *actEntry) bool {
msg := "instance restarting (carlos restart)"
if e.worker {
msg = "sidecar restarting (carlos restart)"
@@ -1390,6 +1396,119 @@ func (a *Service) Bounce(host string) bool {
return true
}
+// BounceAdopted is Bounce for an ADOPT pass's restart callback, with one
+// thing Bounce cannot know taken into account: the callback fires after the
+// pass has repointed live/<host>, so a child already exec'd from the build the
+// symlink names has nothing to pick up, and stopping it buys a drain and a
+// wake for no change of binary.
+//
+// That child exists whenever something else finished the adoption first. A
+// root pass adopts, ConvergeAdopted cycles the instance, a request wakes it
+// onto the new build — and a slower pass that read the old symlink before all
+// of that reaches its own restart afterwards. Bounce would stop the
+// replacement; this leaves it alone.
+//
+// Everything else is Bounce: same no-op on a host that is not awake, same
+// event, same stop, and it applies to a worker exactly as Bounce does. An
+// unknown spawn version or an unreadable symlink bounces — with nothing to
+// compare, the safe reading of "adopted" is "the child may be stale".
+//
+// `carlos restart` and config delivery keep the unconditional Bounce: there
+// the request IS the reason, whatever build is running.
+func (a *Service) BounceAdopted(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.state != actAwake {
+ return false
+ }
+ if e.spawnVersion != "" {
+ if cur, err := a.liveVer(host); err == nil && cur == e.spawnVersion {
+ return false
+ }
+ }
+ return a.bounceLocked(e)
+}
+
+// Drifted lists the awake INSTANCES whose child was exec'd from a build
+// live/<host> no longer names, each with the version the symlink names now.
+// It is the cheap half of ConvergeAdopted below: a caller polling at seconds
+// cadence asks this first and, on the ordinary tick where nothing has shipped,
+// gets nothing back and reads no registry row at all. One readlink per awake
+// instance, the same read the sweep makes.
+//
+// A read error, an unknown spawn version and an unknown symlink target are all
+// "not drifted" — the sweep's own rule (ConvergeDecision): with nothing to
+// compare there is nothing to act on. Workers are never listed; see
+// ConvergeAdopted for why they stay with the sweep.
+func (a *Service) Drifted() map[string]string {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ var out map[string]string
+ for _, e := range a.entries {
+ if e.worker || e.state != actAwake || e.spawnVersion == "" || a.fences[e.host] != 0 {
+ continue
+ }
+ cur, err := a.liveVer(e.host)
+ if err != nil || cur == "" || cur == e.spawnVersion {
+ continue
+ }
+ if out == nil {
+ out = map[string]string{}
+ }
+ out[e.host] = cur
+ }
+ return out
+}
+
+// ConvergeAdopted cycles host's awake instance onto ver, a version an adopt
+// pass has FINISHED adopting for it, without the sweep's ConvergeGrace. It
+// reports whether it started a cycle.
+//
+// It exists because the adoption a real box performs is delegated. The agent
+// runs sandboxed and cannot write the version store, so the root pass stages,
+// repoints live/<host> and stamps the route — and cycles no exec child,
+// because the children belong to this process. The agent's own pass then
+// finds the host up to date and never reaches its restart callback, which
+// left the sweep's grace as the only thing standing between a deploy and the
+// build it shipped (discussion #231). This is the restart half of that
+// adoption, performed by the one process that can.
+//
+// The grace is safe to skip HERE and nowhere else, because of what the caller
+// vouches for: ver is the version the route is STAMPED with, and the stamp is
+// written only after the adopting pass is done with the host — past its own
+// restart and any rollback. The sweep cannot know that; it sees a symlink
+// move and waits to learn whether it holds. ver must also still be what the
+// symlink names, checked here under the lock: a stamp the symlink has moved
+// on from is a newer adoption in flight, and its own stamp will bring the
+// caller back.
+//
+// Like Bounce it never starts anything — an instance's next request is the
+// wake half — and it is the same stop the sweep's convergence performs:
+// SIGTERM drain, final checkpoint, lease release.
+//
+// Instances only. A worker's registry version is what its child REPORTS
+// (cleared on every respawn), not what adoption stamped, so the agreement
+// this rests on does not exist for one; the sweep's convergeWorkerLocked,
+// behind the grace, remains a worker's path onto a new build.
+func (a *Service) ConvergeAdopted(host, ver string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.worker || e.state != actAwake || a.fences[host] != 0 {
+ return false
+ }
+ if ver == "" || e.spawnVersion == "" || e.spawnVersion == ver {
+ return false
+ }
+ if cur, err := a.liveVer(host); err != nil || cur != ver {
+ return false
+ }
+ log.Printf("activator: %s serving %s but %s is adopted — cycling onto the new build", host, e.spawnVersion, ver)
+ a.beginStopLocked(e)
+ return true
+}
+
// --- standalone sidecars (workers) -------------------------------------------
//
// A worker is a member of this same state machine with the request path
diff --git a/internal/activator/converge_adopted_test.go b/internal/activator/converge_adopted_test.go
new file mode 100644
index 000000000..31f5781e9
--- /dev/null
+++ b/internal/activator/converge_adopted_test.go
@@ -0,0 +1,203 @@
+package activator
+
+import (
+ "context"
+ "reflect"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// The adoption a real box performs is delegated: the sandboxed agent cannot
+// write the version store, so the ROOT pass stages, repoints live/<host> and
+// stamps the version — and cycles no exec child, because the children are the
+// agent's. Nothing then told the agent the symlink had moved, and the awake
+// child served the old build until the sweep's grace ran out (discussion #231:
+// 110-200s per deploy on the flagship). ConvergeAdopted is the agent finishing
+// the half it delegated: no grace, no sweep.
+func TestConvergeAdoptedCyclesAStaleInstanceWithoutTheGrace(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour) // idle far away — only convergence can stop it
+ var ver atomic.Value
+ ver.Store("v1")
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ b.execVerFn = func() string { return ver.Load().(string) }
+ host := "one.example.com"
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v with the symlink unmoved, want none", got)
+ }
+
+ ver.Store("v2") // the root pass repoints under the running child
+ if got, want := a.Drifted(), map[string]string{host: "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("Drifted = %v, want %v", got, want)
+ }
+ if !a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted refused an awake instance serving v1 under a symlink that names v2")
+ }
+ // No clock advance and no sweep: the cycle is the call's own doing.
+ waitFor(t, "the adopted-version cycle", func() bool { return a.entryCount() == 0 })
+ if !b.lastProc().isStopped() {
+ t.Fatal("stale instance's process was not stopped")
+ }
+ if !b.lastRep().isClosed() {
+ t.Fatal("the cycle skipped the final checkpoint / lease release")
+ }
+
+ // The next request wakes it onto v2, and a converged child is left alone.
+ release, err = a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("re-wake after the cycle: %v", err)
+ }
+ release()
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v after re-waking onto v2, want none", got)
+ }
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted cycled a child that already runs the adopted version")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("a converged instance was stopped")
+ }
+}
+
+// Every refusal leaves the instance exactly as it was. In particular it never
+// WAKES anything: a sleeper picks the new build up at its own next wake.
+func TestConvergeAdoptedRefusals(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ var ver atomic.Value
+ ver.Store("v1")
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ b.execVerFn = func() string { return ver.Load().(string) }
+ host := "one.example.com"
+
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted acted on a host that is asleep")
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("ConvergeAdopted woke a sleeping host")
+ }
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ ver.Store("v2")
+
+ // The caller's version and the symlink must agree. A stamp that names
+ // something the symlink does not is an adoption still in flight (or
+ // already superseded), and cycling now would land on whichever build the
+ // symlink happens to name.
+ if a.ConvergeAdopted(host, "v3") {
+ t.Fatal("ConvergeAdopted cycled onto a version the live symlink does not name")
+ }
+ if a.ConvergeAdopted(host, "") {
+ t.Fatal("ConvergeAdopted cycled for an empty version")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("a refused ConvergeAdopted stopped the instance")
+ }
+}
+
+// An unknown spawn version is "don't converge", the same rule the sweep keeps:
+// with nothing to compare, a cycle could never be shown to have been needed.
+func TestConvergeAdoptedLeavesAnUnknownSpawnVersionAlone(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.liveVer = func(string) (string, error) { return "v2", nil }
+ b.execVerFn = func() string { return "" }
+ host := "one.example.com"
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v for a child whose spawn version is unknown, want none", got)
+ }
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted cycled a child whose spawn version is unknown")
+ }
+}
+
+// A worker is out of this path's reach on purpose: its registry version is
+// what the child REPORTS, not what adoption stamped, so "the stamp agrees with
+// the symlink" means something else there. The sweep's own worker converge
+// (one move, behind the grace) still covers it.
+func TestConvergeAdoptedLeavesWorkersToTheSweep(t *testing.T) {
+ a, _, _, ver := shippingActivator(t, time.Hour)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ waitFor(t, "the worker to be awake", func() bool { return a.entrySpawnVersion(host) == "v1" })
+ ver.Store("v2")
+ if got := a.Drifted(); len(got) != 0 {
+ t.Fatalf("Drifted = %v, want no workers in it", got)
+ }
+ if a.ConvergeAdopted(host, "v2") {
+ t.Fatal("ConvergeAdopted cycled a worker")
+ }
+}
+
+// An adopt pass's restart callback fires after it has repointed live/<host>,
+// so by then the symlink names the version being adopted. A child ALREADY
+// exec'd from that version has nothing to pick up: stopping it is one more
+// drain and wake for no change of build. That child exists whenever something
+// else got there first — ConvergeAdopted finishing a root pass's adoption and
+// a request re-waking the instance, while a slower pass was still mid-host.
+func TestBounceAdoptedSparesAChildAlreadyOnTheLiveBuild(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ var ver atomic.Value
+ ver.Store("v2")
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ b.execVerFn = func() string { return ver.Load().(string) }
+ host := "one.example.com"
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted stopped a child that already runs the build the symlink names")
+ }
+ if a.entryCount() != 1 {
+ t.Fatal("the converged child was stopped")
+ }
+
+ // The ordinary case is unchanged: the symlink has moved under the child.
+ ver.Store("v3")
+ if !a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted spared a child serving v2 under a symlink that names v3")
+ }
+ waitFor(t, "the adoption bounce", func() bool { return a.entryCount() == 0 })
+ if a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted acted on a host that is asleep")
+ }
+}
+
+// With nothing to compare it bounces, exactly as Bounce always has: a backend
+// that cannot name its build, or a symlink that cannot be read, must not turn
+// an adoption into a silent no-op over a child still on the old binary.
+func TestBounceAdoptedBouncesWhenTheBuildIsUnknown(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.liveVer = func(string) (string, error) { return "v2", nil }
+ b.execVerFn = func() string { return "" }
+ host := "one.example.com"
+
+ release, err := a.beginRequest(context.Background(), testRoute(host))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+ if !a.BounceAdopted(host) {
+ t.Fatal("BounceAdopted refused a child whose spawn version is unknown")
+ }
+ waitFor(t, "the bounce", func() bool { return a.entryCount() == 0 })
+}
diff --git a/internal/edge/static.go b/internal/edge/static.go
index ea43c68ba..247ca200d 100644
--- a/internal/edge/static.go
+++ b/internal/edge/static.go
@@ -17,16 +17,20 @@ import (
"crypto/ed25519"
"crypto/sha256"
"encoding/hex"
+ "errors"
"fmt"
"io"
"io/fs"
"log"
+ "mime"
"net/http"
"os"
"path"
"path/filepath"
"sort"
+ "strings"
"sync"
+ "sync/atomic"
"time"
"github.com/carlosframework/platform/internal/apperr"
@@ -163,6 +167,18 @@ func (s *staticSite) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Set before any body is written, so it lands on the tree's own 404s too
// (the handler serving that tree IS a version).
w.Header().Set("X-Carlos-Version", version)
+ // A page out of the tree is revalidated before it is reused, and so is
+ // the tree's own 404. With Last-Modified alone a browser GUESSES a
+ // freshness lifetime — typically a tenth of the file's age — and reuses
+ // its copy without asking, so a site last deployed a month ago kept
+ // showing the old page for days after the next deploy (discussion #195).
+ // no-cache is "store it, but ask first"; the ETag serveTree sets is what
+ // makes asking cost a 304. It also keeps these responses out of the
+ // edge's own shared cache, which admits only a positive lifetime — as
+ // they always were, having named none. (net/http strips the header again
+ // from the errors it writes itself, a 416 or a 400: neither is a status a
+ // browser caches on a guess.)
+ w.Header().Set("Cache-Control", "no-cache")
s.serveTree(dir, w, r)
}
@@ -363,7 +379,26 @@ func extractSiteTar(data []byte, dst string) error {
// serveTree answers one request from an extracted version dir:
// http.FileServer semantics with index.html for directories and a clean 404
// (never a directory listing) when neither the file nor an index exists.
+//
+// It serves the file itself rather than through http.ServeFile, because
+// ServeFile offers the file's mtime as Last-Modified and answers
+// If-Modified-Since from it — and an extracted file's mtime is when THIS box
+// unpacked the tree, which is not a fact about the content. It differs
+// between edges serving the same release, and it runs backwards on a
+// rollback: the tree rolled back to was unpacked before the one the browser
+// holds, so "modified since?" answers no and the browser keeps the build that
+// was withdrawn. No date is offered, so there is none to compare.
+//
+// The validator is the file's content (tagCache). What ServeFile did besides
+// is kept by hand: the refusal of a ".." element, and the redirect off a
+// trailing /index.html onto its directory.
func (s *staticSite) serveTree(dir string, w http.ResponseWriter, r *http.Request) {
+ if containsDotDot(r.URL.Path) {
+ // The Clean below would resolve it harmlessly inside the tree; this
+ // keeps the answer ServeFile always gave such a path.
+ http.Error(w, "invalid URL path", http.StatusBadRequest)
+ return
+ }
// Clean rooted at "/" so no request path can climb out of the tree.
p := path.Clean("/" + r.URL.Path)
fp := filepath.Join(dir, filepath.FromSlash(p))
@@ -376,6 +411,270 @@ func (s *staticSite) serveTree(dir string, w http.ResponseWriter, r *http.Reques
http.NotFound(w, r)
return
}
- // ServeFile brings Content-Type, ranges, If-Modified-Since for free.
- http.ServeFile(w, r, fp)
+ if strings.HasSuffix(r.URL.Path, "/index.html") {
+ // Relative, as ServeFile wrote it, so it survives any prefix the
+ // request arrived under.
+ loc := "./"
+ if q := r.URL.RawQuery; q != "" {
+ loc += "?" + q
+ }
+ w.Header().Set("Location", loc)
+ w.WriteHeader(http.StatusMovedPermanently)
+ return
+ }
+ f, err := os.Open(fp)
+ if err != nil {
+ // ServeFile's mapping, kept: a file that vanished since the Stat is
+ // a 404, but one that is there and will not open is not a missing
+ // page, and must not read as one in an access log.
+ switch {
+ case errors.Is(err, fs.ErrNotExist):
+ http.NotFound(w, r)
+ case errors.Is(err, fs.ErrPermission):
+ http.Error(w, "403 Forbidden", http.StatusForbidden)
+ default:
+ log.Printf("edge: static %s/%s: open %s: %v", s.app, s.channel, p, err)
+ http.Error(w, "500 Internal Server Error", http.StatusInternalServerError)
+ }
+ return
+ }
+ defer f.Close()
+ // Stat the OPEN file: its identity is what a remembered tag is checked
+ // against, and it is the file whose bytes are about to be sent.
+ if ofi, err := f.Stat(); err == nil && !ofi.IsDir() {
+ fi = ofi
+ if tag, ctype, err := staticTags.tag(r.Context(), fp, f, fi); err == nil {
+ // The type the tag was computed over is the type sent: set here,
+ // ServeContent leaves it alone.
+ w.Header().Set("ETag", tag)
+ w.Header().Set("Content-Type", ctype)
+ } else {
+ if !errors.Is(err, errTagBusy) && r.Context().Err() == nil {
+ // No tag is the honest answer: the response then carries no
+ // validator and no-cache makes the browser refetch it whole.
+ log.Printf("edge: static %s/%s: tag %s: %v", s.app, s.channel, p, err)
+ }
+ // A failed hash rewinds, but if even that failed the file's
+ // position is unknown and ServeContent would sniff its type
+ // from the middle of it.
+ if _, serr := f.Seek(0, io.SeekStart); serr != nil {
+ log.Printf("edge: static %s/%s: rewind %s: %v", s.app, s.channel, p, serr)
+ http.Error(w, "500 Internal Server Error", http.StatusInternalServerError)
+ return
+ }
+ }
+ }
+ // ServeContent brings Content-Type, ranges and the conditional-request
+ // handling; the zero time is "no Last-Modified, and no date to test".
+ http.ServeContent(w, r, fi.Name(), time.Time{}, f)
+}
+
+// tagCache computes and remembers the entity tag of each extracted file: a
+// hash of its bytes and of the one thing besides them that decides what the
+// client receives, its Content-Type.
+//
+// The content, and nothing that merely names it. A version name is reused —
+// a pruned version re-shipped under its old name is a new release
+// (release.putImmutable), and a hostname can move between two apps that both
+// have a "v1" — so a tag made of the name would hand a browser a 304 over
+// bytes it does not hold. Hashing also means identical bytes get one tag on
+// every edge, and a file a deploy did not touch keeps its tag across it: the
+// release that changed one page does not make every browser refetch every
+// asset.
+//
+// The type is hashed in because the same bytes can be two representations.
+// /about served from an extensionless file is sniffed as text/plain; the same
+// bytes at about/index.html are text/html; and two boxes whose MIME tables
+// differ type one .md file differently. A 304 carries no Content-Type, so
+// bytes alone would leave the browser rendering the old one. The type is
+// resolved here exactly as net/http resolves it — the extension's mapping,
+// else a sniff of the first 512 bytes — and handed back with the tag, so the
+// type the tag covers is the type the response carries.
+//
+// WEAK, although such a hash is as strong as a validator gets. The edge gzips
+// compressible responses above this handler and weakens a strong tag when it
+// does (compress.go, weakenETag) — but only on the 200. A 304 and a HEAD pass
+// through as they are, so a strong tag would come back weak on the page and
+// strong on its own revalidation, and a cache may not update the one from the
+// other. Weak from the start, it reads the same on all three. The cost is
+// that If-Range falls back to the whole file.
+//
+// One cache for the process rather than one per handler: the edge rebuilds
+// every static handler on each route-table reload, and a per-handler memo
+// would re-hash every site each time a route anywhere on the box changed.
+// What makes a shared, long-lived memo safe:
+//
+// - An entry is checked against the file it was computed FROM, not the path
+// it was computed at (os.SameFile, size, mtime). A tree replaced on disk
+// under a live handler — cache maintenance, then a re-shipped version of
+// the same name extracted again — is new files, and hashes again.
+// - A cold file is hashed once however many requests want it: the rest
+// wait for the one in flight.
+// - A file over tagInline hashes under a semaphore, tagHashers at a time,
+// so a sweep of a site's large assets cannot take every core and the disk
+// at once. Small files never queue behind them: their work is bounded by
+// their size and they hash inline.
+// - Nobody waits long. A tag is an optimization, and a request that has
+// waited `wait` for a hasher or for someone else's hash is served without
+// one (errTagBusy): correct, uncached, and tried again on the next
+// request. Only the request doing a hash sees it through, because every
+// later request needs the result.
+// - It is bounded: past tagCacheMax entries it is emptied and refills on
+// demand, which costs hashing and never correctness.
+type tagCache struct {
+ mu sync.Mutex
+ entries map[string]tagEntry
+ flights map[string]chan struct{} // closed when the hash for that path is done
+ sem chan struct{}
+ wait time.Duration
+
+ // Evidence and a seam for the tests: files hashed, callers currently
+ // waiting on another's hash, and a hook run inside a hash (nil in
+ // production) so a test can hold one open.
+ hashed atomic.Int64
+ waiting atomic.Int64
+ hold func()
+}
+
+type tagEntry struct {
+ fi os.FileInfo
+ tag, ctype string
+}
+
+const (
+ tagHashers = 4
+ tagInline = 1 << 20 // files up to this size hash without a hasher slot
+ tagWait = 250 * time.Millisecond
+ tagCacheMax = 1 << 16
+)
+
+// errTagBusy: the tag could not be had within the wait budget. The caller
+// serves the file untagged.
+var errTagBusy = errors.New("static tag: hashers busy")
+
+func newTagCache() *tagCache {
+ return &tagCache{
+ entries: map[string]tagEntry{},
+ flights: map[string]chan struct{}{},
+ sem: make(chan struct{}, tagHashers),
+ wait: tagWait,
+ }
+}
+
+var staticTags = newTagCache()
+
+// current reports whether e still describes the file fi is.
+func (e tagEntry) current(fi os.FileInfo) bool {
+ return e.fi != nil && os.SameFile(e.fi, fi) && e.fi.Size() == fi.Size() && e.fi.ModTime().Equal(fi.ModTime())
+}
+
+// tag returns the entity tag and Content-Type for the open file f at fp,
+// hashing it if no current entry exists. f is left positioned at its start.
+// ctx and the wait budget bound the WAITING — on another request's hash, or
+// on a free hasher — never a hash in progress.
+func (c *tagCache) tag(ctx context.Context, fp string, f io.ReadSeeker, fi os.FileInfo) (tag, ctype string, err error) {
+ var budget <-chan time.Time
+ for {
+ c.mu.Lock()
+ if e, ok := c.entries[fp]; ok && e.current(fi) {
+ c.mu.Unlock()
+ return e.tag, e.ctype, nil
+ }
+ if budget == nil {
+ t := time.NewTimer(c.wait)
+ defer t.Stop()
+ budget = t.C
+ }
+ if flight, ok := c.flights[fp]; ok {
+ c.mu.Unlock()
+ c.waiting.Add(1)
+ select {
+ case <-flight:
+ c.waiting.Add(-1)
+ continue // look again: usually a hit, a miss if the file differs
+ case <-ctx.Done():
+ c.waiting.Add(-1)
+ return "", "", ctx.Err()
+ case <-budget:
+ c.waiting.Add(-1)
+ return "", "", errTagBusy
+ }
+ }
+ done := make(chan struct{})
+ c.flights[fp] = done
+ c.mu.Unlock()
+
+ tag, ctype, err = c.hash(ctx, budget, f, fi)
+ c.mu.Lock()
+ if err == nil {
+ if len(c.entries) >= tagCacheMax {
+ c.entries = map[string]tagEntry{}
+ }
+ c.entries[fp] = tagEntry{fi: fi, tag: tag, ctype: ctype}
+ }
+ delete(c.flights, fp)
+ c.mu.Unlock()
+ close(done)
+ return tag, ctype, err
+ }
+}
+
+func (c *tagCache) hash(ctx context.Context, budget <-chan time.Time, f io.ReadSeeker, fi os.FileInfo) (tag, ctype string, err error) {
+ if fi.Size() > tagInline {
+ select {
+ case c.sem <- struct{}{}:
+ defer func() { <-c.sem }()
+ case <-ctx.Done():
+ return "", "", ctx.Err()
+ case <-budget:
+ return "", "", errTagBusy
+ }
+ }
+ c.hashed.Add(1)
+ if c.hold != nil {
+ c.hold()
+ }
+ // From here the file is being read, and it goes back to its start on
+ // EVERY path out — a failed hash included. The caller serves the file
+ // either way, and net/http sniffs a Content-Type from wherever the file
+ // is positioned.
+ defer func() {
+ if _, serr := f.Seek(0, io.SeekStart); serr != nil && err == nil {
+ tag, ctype, err = "", "", serr
+ }
+ }()
+ // net/http's own rule (serveContent): the extension's type, else a sniff
+ // of the first 512 bytes.
+ var head [512]byte
+ n, err := io.ReadFull(f, head[:])
+ if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
+ return "", "", err
+ }
+ ctype = mime.TypeByExtension(filepath.Ext(fi.Name()))
+ if ctype == "" {
+ ctype = http.DetectContentType(head[:n])
+ }
+ h := sha256.New()
+ io.WriteString(h, ctype)
+ h.Write([]byte{0})
+ h.Write(head[:n])
+ if _, err := io.Copy(h, f); err != nil {
+ return "", "", err
+ }
+ return `W/"` + hex.EncodeToString(h.Sum(nil)[:16]) + `"`, ctype, nil
+}
+
+// containsDotDot reports whether a URL path has a ".." element — net/http's
+// own test, which http.ServeFile applied before this handler served files
+// itself.
+func containsDotDot(p string) bool {
+ if !strings.Contains(p, "..") {
+ return false
+ }
+ for _, el := range strings.FieldsFunc(p, func(r rune) bool { return r == '/' || r == '\\' }) {
+ if el == ".." {
+ return true
+ }
+ }
+ return false
}
diff --git a/internal/edge/static_cache_test.go b/internal/edge/static_cache_test.go
new file mode 100644
index 000000000..ff7d9178f
--- /dev/null
+++ b/internal/edge/static_cache_test.go
@@ -0,0 +1,596 @@
+package edge
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "mime"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// reqStatic is one request the way a browser holding a cached copy makes it:
+// no redirect following, no transparent gzip (an explicit Accept-Encoding
+// turns the client's own off), and the validators in hdr.
+func reqStatic(t *testing.T, method, base, path string, hdr map[string]string) (*http.Response, string) {
+ t.Helper()
+ req, err := http.NewRequest(method, base+path, nil)
+ if err != nil {
+ t.Fatalf("new request: %v", err)
+ }
+ req.Host = "site.example.com"
+ for k, v := range hdr {
+ req.Header.Set(k, v)
+ }
+ client := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+ resp, err := client.Do(req)
+ if err != nil {
+ t.Fatalf("%s %s%s: %v", method, base, path, err)
+ }
+ defer resp.Body.Close()
+ b, err := io.ReadAll(resp.Body)
+ if err != nil {
+ t.Fatalf("%s %s%s: read body: %v", method, base, path, err)
+ }
+ return resp, string(b)
+}
+
+// siteServer ships files as version of app into its own store and serves it
+// from a fresh handler, behind the same compression wrapper the edge applies.
+func siteServer(t *testing.T, app, version string, files map[string]string) *httptest.Server {
+ t.Helper()
+ srv, _ := siteServerTree(t, app, version, files)
+ return srv
+}
+
+// siteServerTree is siteServer that also says where version's tree is
+// extracted, for the tests that reach under the handler.
+func siteServerTree(t *testing.T, app, version string, files map[string]string) (*httptest.Server, string) {
+ t.Helper()
+ store := release.NewDirStore(filepath.Join(t.TempDir(), "store"))
+ priv, pub := testKeys(t)
+ shipSite(t, store, app, version, files)
+ if err := release.Promote(context.Background(), store, priv, release.OpsAccount, app, "canary/x", version); err != nil {
+ t.Fatalf("Promote %s: %v", version, err)
+ }
+ rt := registry.Route{Host: "site.example.com", Kind: "static", App: app, Channel: "canary/x", Status: "ready"}
+ h := newStaticHandler(rt, store, pub, filepath.Join(t.TempDir(), "cache"))
+ h.(*staticSite).recheck = 0
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if gw := newGzipWriter(w, r); gw != nil {
+ w = gw
+ defer gw.finish()
+ }
+ h.ServeHTTP(w, r)
+ }))
+ t.Cleanup(srv.Close)
+ return srv, h.(*staticSite).dir(version)
+}
+
+// A static response used to carry Last-Modified and nothing else, and a
+// browser handed only that guesses a freshness lifetime — typically a tenth of
+// the file's age — and reuses its copy without asking. A site last deployed a
+// month ago kept showing the old page for days after the next deploy
+// (discussion #195). no-cache makes every load ask; the ETag makes the answer
+// a cheap 304 until the bytes actually change.
+func TestStaticResponsesMakeTheBrowserRevalidate(t *testing.T) {
+ _, _, _, srv := staticFixture(t)
+
+ for _, path := range []string{"/", "/docs/page.html", "/style.css"} {
+ resp, _ := reqStatic(t, "GET", srv.URL, path, nil)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("GET %s = %d, want 200", path, resp.StatusCode)
+ }
+ if got := resp.Header.Get("Cache-Control"); got != "no-cache" {
+ t.Errorf("GET %s Cache-Control = %q, want %q", path, got, "no-cache")
+ }
+ if got := resp.Header.Get("ETag"); !strings.HasPrefix(got, `W/"`) || len(got) < 20 {
+ t.Errorf("GET %s ETag = %q, want a weak content tag", path, got)
+ }
+ // The mtime of an extracted file is when THIS box unpacked it. It is
+ // not a fact about the content and must not be offered as one.
+ if got := resp.Header.Get("Last-Modified"); got != "" {
+ t.Errorf("GET %s Last-Modified = %q, want none", path, got)
+ }
+ }
+
+ // The tree's own 404 must not be cached on a guess either: the page it
+ // stands in for may exist in the next deploy. It names no representation,
+ // so it carries no validator.
+ resp, _ := reqStatic(t, "GET", srv.URL, "/missing.html", nil)
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("GET /missing.html = %d, want 404", resp.StatusCode)
+ }
+ if got := resp.Header.Get("Cache-Control"); got != "no-cache" {
+ t.Errorf("404 Cache-Control = %q, want %q", got, "no-cache")
+ }
+ if got := resp.Header.Get("ETag"); got != "" {
+ t.Errorf("404 ETag = %q, want none", got)
+ }
+}
+
+// The validator is the file's CONTENT. Unchanged bytes revalidate to a 304 —
+// across a deploy too, so a release that touched one page does not make every
+// browser refetch every asset — and changed bytes answer 200, in both
+// directions: a rollback is as much a change as a deploy.
+func TestStaticValidatorFollowsTheContent(t *testing.T) {
+ store, _, priv, srv := staticFixture(t)
+ promote := func(ver string) {
+ t.Helper()
+ if err := release.Promote(context.Background(), store, priv, release.OpsAccount, "site", "canary/x", ver); err != nil {
+ t.Fatalf("Promote %s: %v", ver, err)
+ }
+ }
+ tagOf := func(path string) string {
+ t.Helper()
+ resp, _ := reqStatic(t, "GET", srv.URL, path, nil)
+ tag := resp.Header.Get("ETag")
+ if resp.StatusCode != http.StatusOK || tag == "" {
+ t.Fatalf("GET %s = %d with ETag %q", path, resp.StatusCode, tag)
+ }
+ return tag
+ }
+ homeV1, cssV1 := tagOf("/"), tagOf("/style.css")
+ if homeV1 == cssV1 {
+ t.Fatalf("two different files share the tag %q", homeV1)
+ }
+
+ resp, _ := reqStatic(t, "GET", srv.URL, "/", map[string]string{"If-None-Match": homeV1})
+ if resp.StatusCode != http.StatusNotModified {
+ t.Fatalf("revalidating unchanged content = %d, want 304", resp.StatusCode)
+ }
+ // A 304 carries what its 200 would have: the cache updates from it.
+ if got := resp.Header.Get("ETag"); got != homeV1 {
+ t.Errorf("304 ETag = %q, want the 200's %q", got, homeV1)
+ }
+ if got := resp.Header.Get("Cache-Control"); got != "no-cache" {
+ t.Errorf("304 Cache-Control = %q, want %q", got, "no-cache")
+ }
+
+ // v2 changes the home page and leaves the stylesheet byte-identical.
+ shipSite(t, store, "site", "v2", map[string]string{"index.html": "home v2", "style.css": "body{}"})
+ promote("v2")
+ resp, body := reqStatic(t, "GET", srv.URL, "/", map[string]string{"If-None-Match": homeV1})
+ if resp.StatusCode != http.StatusOK || body != "home v2" {
+ t.Fatalf("revalidating a changed page after a deploy = %d %q, want 200 %q", resp.StatusCode, body, "home v2")
+ }
+ homeV2 := resp.Header.Get("ETag")
+ if homeV2 == homeV1 {
+ t.Fatalf("the changed page kept the tag %q", homeV1)
+ }
+ resp, _ = reqStatic(t, "GET", srv.URL, "/style.css", map[string]string{"If-None-Match": cssV1})
+ if resp.StatusCode != http.StatusNotModified {
+ t.Fatalf("revalidating an UNCHANGED file after a deploy = %d, want 304", resp.StatusCode)
+ }
+
+ promote("v1") // roll back
+ resp, body = reqStatic(t, "GET", srv.URL, "/", map[string]string{"If-None-Match": homeV2})
+ if resp.StatusCode != http.StatusOK || body != "home v1" {
+ t.Fatalf("revalidating after a rollback = %d %q, want 200 %q", resp.StatusCode, body, "home v1")
+ }
+ if got := resp.Header.Get("ETag"); got != homeV1 {
+ t.Errorf("the rolled-back page's tag = %q, want its original %q", got, homeV1)
+ }
+}
+
+// A browser whose copy predates the ETag holds only the old Last-Modified and
+// revalidates on the date alone. Against an extraction mtime that answer is
+// wrong after a rollback — the tree rolled back to was unpacked BEFORE the one
+// the browser holds, so "modified since?" says no. With no date offered there
+// is nothing to compare, and the browser gets the bytes.
+func TestStaticDateOnlyRevalidationGetsTheBytes(t *testing.T) {
+ _, _, _, srv := staticFixture(t)
+ resp, body := reqStatic(t, "GET", srv.URL, "/", map[string]string{
+ "If-Modified-Since": "Fri, 01 Jan 2100 00:00:00 GMT",
+ })
+ if resp.StatusCode != http.StatusOK || body != "home v1" {
+ t.Fatalf("date-only revalidation = %d %q, want 200 %q", resp.StatusCode, body, "home v1")
+ }
+}
+
+// A version NAME is not a fact about the bytes. A pruned version can be
+// re-shipped under its old name with different content, and a hostname can
+// move between two apps that both have a "v1" — in either case a tag made of
+// the name would let a browser keep the old page on a 304.
+func TestStaticSameVersionNameDifferentContent(t *testing.T) {
+ a := siteServer(t, "site", "v1", map[string]string{"index.html": "the first site"})
+ b := siteServer(t, "other", "v1", map[string]string{"index.html": "a different site"})
+
+ respA, _ := reqStatic(t, "GET", a.URL, "/", nil)
+ tagA := respA.Header.Get("ETag")
+ if tagA == "" {
+ t.Fatal("no ETag from the first site")
+ }
+ resp, body := reqStatic(t, "GET", b.URL, "/", map[string]string{"If-None-Match": tagA})
+ if resp.StatusCode != http.StatusOK || body != "a different site" {
+ t.Fatalf("revalidating one v1 against another app's v1 = %d %q, want 200 %q", resp.StatusCode, body, "a different site")
+ }
+
+ // And identical bytes agree wherever they are served from — two edges
+ // unpacking the same release must hand out the same tag.
+ c := siteServer(t, "third", "v9", map[string]string{"index.html": "the first site"})
+ resp, _ = reqStatic(t, "GET", c.URL, "/", map[string]string{"If-None-Match": tagA})
+ if resp.StatusCode != http.StatusNotModified {
+ t.Fatalf("identical bytes from another handler = %d, want 304", resp.StatusCode)
+ }
+}
+
+// The edge gzips a compressible response and, because gzipped bytes are a
+// different representation, weakens a strong ETag as it does — but only on
+// the 200. A 304 and a HEAD pass through untouched, so a strong tag would
+// come back weak on the page and strong on its revalidation, and a cache may
+// not update the one from the other. A tag that is weak from the start reads
+// the same on all three.
+func TestStaticTagIsTheSameCompressedRevalidatedAndOnHead(t *testing.T) {
+ page := "<html>" + strings.Repeat("a page big enough to be worth compressing. ", 200) + "</html>"
+ srv := siteServer(t, "site", "v1", map[string]string{"index.html": page})
+ gz := map[string]string{"Accept-Encoding": "gzip"}
+
+ resp, _ := reqStatic(t, "GET", srv.URL, "/", gz)
+ if resp.StatusCode != http.StatusOK || resp.Header.Get("Content-Encoding") != "gzip" {
+ t.Fatalf("GET / = %d, Content-Encoding %q; want a gzipped 200", resp.StatusCode, resp.Header.Get("Content-Encoding"))
+ }
+ tag := resp.Header.Get("ETag")
+ if !strings.HasPrefix(tag, `W/"`) {
+ t.Fatalf("gzipped 200 ETag = %q, want a weak tag", tag)
+ }
+
+ resp, _ = reqStatic(t, "GET", srv.URL, "/", map[string]string{"Accept-Encoding": "gzip", "If-None-Match": tag})
+ if resp.StatusCode != http.StatusNotModified {
+ t.Fatalf("revalidating the gzipped page = %d, want 304", resp.StatusCode)
+ }
+ if got := resp.Header.Get("ETag"); got != tag {
+ t.Errorf("304 ETag = %q, want the 200's %q", got, tag)
+ }
+
+ resp, _ = reqStatic(t, "HEAD", srv.URL, "/", gz)
+ if got := resp.Header.Get("ETag"); got != tag {
+ t.Errorf("HEAD ETag = %q, want the GET's %q", got, tag)
+ }
+ // Asked for by name: with no Accept-Encoding of its own the Go client
+ // requests gzip and hides that it did.
+ resp, body := reqStatic(t, "GET", srv.URL, "/", map[string]string{"Accept-Encoding": "identity"})
+ if resp.StatusCode != http.StatusOK || body != page || resp.Header.Get("Content-Encoding") != "" {
+ t.Fatalf("identity GET = %d, Content-Encoding %q, %d bytes; want the plain page", resp.StatusCode, resp.Header.Get("Content-Encoding"), len(body))
+ }
+ if got := resp.Header.Get("ETag"); got != tag {
+ t.Errorf("identity GET ETag = %q, want the gzipped GET's %q", got, tag)
+ }
+}
+
+// What http.ServeFile gave for free and this handler must keep now that it
+// serves the file itself: the canonical redirect off /index.html, ranges, and
+// the refusal of a path with a ".." element.
+func TestStaticKeepsServeFileBehaviour(t *testing.T) {
+ _, _, _, srv := staticFixture(t)
+
+ resp, _ := reqStatic(t, "GET", srv.URL, "/index.html?x=1", nil)
+ if resp.StatusCode != http.StatusMovedPermanently || resp.Header.Get("Location") != "./?x=1" {
+ t.Fatalf("GET /index.html = %d, Location %q; want 301 to ./?x=1", resp.StatusCode, resp.Header.Get("Location"))
+ }
+
+ resp, body := reqStatic(t, "GET", srv.URL, "/", map[string]string{"Range": "bytes=0-3"})
+ if resp.StatusCode != http.StatusPartialContent || body != "home" {
+ t.Fatalf("ranged GET = %d %q, want 206 %q", resp.StatusCode, body, "home")
+ }
+ // A weak tag cannot vouch for a byte range: If-Range falls back to the
+ // whole file, which is the safe answer and the documented cost.
+ tag := resp.Header.Get("ETag")
+ resp, body = reqStatic(t, "GET", srv.URL, "/", map[string]string{"Range": "bytes=0-3", "If-Range": tag})
+ if resp.StatusCode != http.StatusOK || body != "home v1" {
+ t.Fatalf("If-Range with a weak tag = %d %q, want the full 200", resp.StatusCode, body)
+ }
+
+ req, err := http.NewRequest("GET", srv.URL+"/", nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Host = "site.example.com"
+ req.URL.Opaque = "/docs/../index.html" // sent as written, not cleaned by the client
+ resp2, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp2.Body.Close()
+ if resp2.StatusCode != http.StatusBadRequest {
+ t.Fatalf("a path with a .. element = %d, want 400", resp2.StatusCode)
+ }
+}
+
+// The same bytes can be two representations. /about as an extensionless file
+// is sniffed as text/plain; redeployed as about/index.html it is text/html,
+// byte for byte the same. A tag made of the bytes alone would answer 304 and
+// leave the browser rendering markup as text.
+func TestStaticTagCoversTheContentType(t *testing.T) {
+ store, _, priv, srv := staticFixture(t)
+ const same = "Hello <em>world</em>"
+ shipSite(t, store, "site", "v2", map[string]string{"index.html": "home", "about": same})
+ if err := release.Promote(context.Background(), store, priv, release.OpsAccount, "site", "canary/x", "v2"); err != nil {
+ t.Fatalf("Promote v2: %v", err)
+ }
+ resp, _ := reqStatic(t, "GET", srv.URL, "/about", nil)
+ plainTag, plainType := resp.Header.Get("ETag"), resp.Header.Get("Content-Type")
+ if resp.StatusCode != http.StatusOK || !strings.HasPrefix(plainType, "text/plain") {
+ t.Fatalf("GET /about = %d %q, want a text/plain 200", resp.StatusCode, plainType)
+ }
+
+ shipSite(t, store, "site", "v3", map[string]string{"index.html": "home", "about/index.html": same})
+ if err := release.Promote(context.Background(), store, priv, release.OpsAccount, "site", "canary/x", "v3"); err != nil {
+ t.Fatalf("Promote v3: %v", err)
+ }
+ resp, body := reqStatic(t, "GET", srv.URL, "/about", map[string]string{"If-None-Match": plainTag})
+ if resp.StatusCode != http.StatusOK || body != same || !strings.HasPrefix(resp.Header.Get("Content-Type"), "text/html") {
+ t.Fatalf("revalidating /about after it became HTML = %d %q; want a text/html 200", resp.StatusCode, resp.Header.Get("Content-Type"))
+ }
+}
+
+// openTagged writes body at dir/name and opens it the way serveTree does.
+func openTagged(t *testing.T, dir, name, body string) (string, *os.File, os.FileInfo) {
+ t.Helper()
+ fp := filepath.Join(dir, name)
+ if err := os.WriteFile(fp, []byte(body), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ f, err := os.Open(fp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { f.Close() })
+ fi, err := f.Stat()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return fp, f, fi
+}
+
+// A cold file is hashed ONCE however many requests arrive for it together.
+// Without that, a burst of requests against one large uncached asset — HEADs
+// and one-byte ranges included — would each read and hash the whole file
+// before sending a byte.
+//
+// The overlap is forced, not hoped for: the first hash is held until every
+// other caller is known to be waiting on it.
+func TestTagCacheHashesAColdFileOnceForOverlappingRequests(t *testing.T) {
+ c := newTagCache()
+ c.wait = time.Minute
+ held, release := make(chan struct{}), make(chan struct{})
+ var heldOnce, releaseOnce sync.Once
+ c.hold = func() { heldOnce.Do(func() { close(held) }); <-release }
+ letGo := func() { releaseOnce.Do(func() { close(release) }) }
+ defer letGo() // a failure below must not strand the callers
+ dir := t.TempDir()
+ if err := os.WriteFile(filepath.Join(dir, "big.bin"), []byte(strings.Repeat("x", 4096)), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ const callers = 12
+ tags := make([]string, callers)
+ errs := make([]error, callers)
+ var wg sync.WaitGroup
+ call := func(i int) {
+ defer wg.Done()
+ fp := filepath.Join(dir, "big.bin")
+ f, err := os.Open(fp)
+ if err != nil {
+ errs[i] = err
+ return
+ }
+ defer f.Close()
+ fi, err := f.Stat()
+ if err != nil {
+ errs[i] = err
+ return
+ }
+ tags[i], _, errs[i] = c.tag(context.Background(), fp, f, fi)
+ }
+ wg.Add(1)
+ go call(0)
+ <-held // the first caller is inside its hash
+ for i := 1; i < callers; i++ {
+ wg.Add(1)
+ go call(i)
+ }
+ deadline := time.Now().Add(10 * time.Second)
+ for c.waiting.Load() != callers-1 {
+ if time.Now().After(deadline) {
+ t.Fatalf("%d of %d callers are waiting on the hash in flight", c.waiting.Load(), callers-1)
+ }
+ time.Sleep(time.Millisecond)
+ }
+ letGo()
+ wg.Wait()
+ for i := range tags {
+ if errs[i] != nil || tags[i] == "" || tags[i] != tags[0] {
+ t.Fatalf("caller %d = %q, %v; want every caller to get %q", i, tags[i], errs[i], tags[0])
+ }
+ }
+ if got := c.hashed.Load(); got != 1 {
+ t.Fatalf("the file was hashed %d times for %d overlapping callers, want once", got, callers)
+ }
+}
+
+// Large files hash a few at a time, and a request that cannot get a turn —
+// or whose file is still being hashed for someone else — is not made to wait
+// for it: past a short budget it is served without a tag. A small file never
+// queues behind them at all. Otherwise four requests for four large cold
+// assets would hold up the first load of every other static site on the box.
+func TestTagCacheDoesNotQueueBehindLargeHashes(t *testing.T) {
+ c := newTagCache()
+ c.wait = 20 * time.Millisecond
+ dir := t.TempDir()
+ for i := 0; i < tagHashers; i++ {
+ c.sem <- struct{}{} // every hasher busy with someone's large asset
+ }
+
+ fp, f, fi := openTagged(t, dir, "big.bin", strings.Repeat("x", tagInline+1))
+ if _, _, err := c.tag(context.Background(), fp, f, fi); !errors.Is(err, errTagBusy) {
+ t.Fatalf("a large file with no free hasher = %v, want errTagBusy", err)
+ }
+ if n := len(c.flights); n != 0 {
+ t.Fatalf("%d flights left registered after giving up", n)
+ }
+
+ fp, f, fi = openTagged(t, dir, "page.html", "<p>a small page</p>")
+ tag, ctype, err := c.tag(context.Background(), fp, f, fi)
+ if err != nil || tag == "" || !strings.HasPrefix(ctype, "text/html") {
+ t.Fatalf("a small file while every hasher is busy = %q %q %v; want it tagged at once", tag, ctype, err)
+ }
+
+ // Once a hasher frees up the large file is tagged, and stays tagged.
+ // The budget is lifted first: at 20ms a stalled scheduler could find the
+ // timer and the free hasher ready together, and pick the timer.
+ <-c.sem
+ c.wait = time.Minute
+ fp, f, fi = openTagged(t, dir, "big.bin", strings.Repeat("x", tagInline+1))
+ if tag, _, err := c.tag(context.Background(), fp, f, fi); err != nil || tag == "" {
+ t.Fatalf("a large file with a free hasher = %q, %v", tag, err)
+ }
+}
+
+// The tag covers the Content-Type the response will carry, not the extension
+// it was derived from: two boxes with different MIME tables type the same
+// bytes differently, and a 304 from one must not confirm the other's page.
+func TestTagCacheTagCoversTheResolvedContentType(t *testing.T) {
+ c := newTagCache()
+ const body = "# notes\n\nsome text\n"
+ // The MIME table is process-wide and an added type stays added, so each
+ // run takes an extension of its own (-count=N would otherwise find the
+ // last run's mapping in place of an unknown one).
+ ext := fmt.Sprintf(".zzcarlos%d", time.Now().UnixNano())
+ fp, f, fi := openTagged(t, t.TempDir(), "notes"+ext, body)
+ sniffedTag, sniffedType, err := c.tag(context.Background(), fp, f, fi)
+ if err != nil || !strings.HasPrefix(sniffedType, "text/plain") {
+ t.Fatalf("an unknown extension = %q %v, want it sniffed as text/plain", sniffedType, err)
+ }
+
+ // "Another box": the same bytes under the same name, with the type known.
+ if err := mime.AddExtensionType(ext, "text/markdown; charset=utf-8"); err != nil {
+ t.Fatal(err)
+ }
+ fp, f, fi = openTagged(t, t.TempDir(), "notes"+ext, body)
+ knownTag, knownType, err := c.tag(context.Background(), fp, f, fi)
+ if err != nil || !strings.HasPrefix(knownType, "text/markdown") {
+ t.Fatalf("a mapped extension = %q %v, want text/markdown", knownType, err)
+ }
+ if knownTag == sniffedTag {
+ t.Fatalf("the same bytes typed %q and %q share the tag %s", sniffedType, knownType, knownTag)
+ }
+}
+
+// A remembered tag describes the file it was computed from, not the path. If
+// the tree is replaced on disk under a live handler — cache maintenance, then
+// another handler unpacking a re-shipped version of the same name — the old
+// tag must not be handed out over the new bytes, nor a 304 against it.
+func TestStaticTagDoesNotOutliveTheFileItDescribes(t *testing.T) {
+ srv, tree := siteServerTree(t, "site", "v1", map[string]string{"index.html": "home", "style.css": "body{}"})
+ resp, _ := reqStatic(t, "GET", srv.URL, "/style.css", nil)
+ oldTag := resp.Header.Get("ETag")
+ if oldTag == "" {
+ t.Fatal("no ETag for /style.css")
+ }
+
+ // Replace the file the way a fresh extraction would — a new file renamed
+ // into place — and make it the hardest case to notice: the same length
+ // and the same mtime, so only the file's identity gives it away.
+ fp := filepath.Join(tree, "style.css")
+ before, err := os.Stat(fp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(fp+".new", []byte("main{}"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chtimes(fp+".new", before.ModTime(), before.ModTime()); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Rename(fp+".new", fp); err != nil {
+ t.Fatal(err)
+ }
+
+ resp, body := reqStatic(t, "GET", srv.URL, "/style.css", map[string]string{"If-None-Match": oldTag})
+ if resp.StatusCode != http.StatusOK || body != "main{}" {
+ t.Fatalf("revalidating against a replaced file = %d %q, want 200 with the new bytes", resp.StatusCode, body)
+ }
+ if got := resp.Header.Get("ETag"); got == "" || got == oldTag {
+ t.Fatalf("ETag over the replaced file = %q, want a new tag (old %q)", got, oldTag)
+ }
+}
+
+// A file that is there and cannot be read is not a missing page. ServeFile
+// said 403, and an operator reading the access log must still be able to tell
+// a broken cache from a mistyped URL.
+func TestStaticUnreadableFileIsNotA404(t *testing.T) {
+ if os.Geteuid() == 0 {
+ t.Skip("root reads through file modes")
+ }
+ srv, tree := siteServerTree(t, "site", "v1", map[string]string{"index.html": "home", "style.css": "body{}"})
+ reqStatic(t, "GET", srv.URL, "/", nil) // extract the tree
+ fp := filepath.Join(tree, "style.css")
+ if err := os.Chmod(fp, 0); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.Chmod(fp, 0o644) })
+ resp, _ := reqStatic(t, "GET", srv.URL, "/style.css", nil)
+ if resp.StatusCode != http.StatusForbidden {
+ t.Fatalf("GET of an unreadable file = %d, want 403", resp.StatusCode)
+ }
+}
+
+// failingFile reads like a file whose later blocks are bad: the first good
+// bytes come back, then every read fails. Seeks work.
+type failingFile struct {
+ data []byte
+ good int
+ pos int64
+}
+
+func (f *failingFile) Read(p []byte) (int, error) {
+ if f.pos >= int64(f.good) {
+ return 0, errors.New("input/output error")
+ }
+ n := copy(p, f.data[f.pos:f.good])
+ f.pos += int64(n)
+ return n, nil
+}
+
+func (f *failingFile) Seek(off int64, whence int) (int64, error) {
+ switch whence {
+ case io.SeekStart:
+ f.pos = off
+ case io.SeekCurrent:
+ f.pos += off
+ case io.SeekEnd:
+ f.pos = int64(len(f.data)) + off
+ }
+ return f.pos, nil
+}
+
+// A hash that fails part-way must hand the file back at its start. The
+// handler then serves it untagged, and net/http sniffs the Content-Type from
+// wherever the file is positioned: left mid-file, an extensionless image
+// would go out as text/plain.
+func TestTagCacheRewindsTheFileWhenHashingFails(t *testing.T) {
+ c := newTagCache()
+ _, _, fi := openTagged(t, t.TempDir(), "picture", "placeholder")
+ for _, good := range []int{100, 2000} { // fails inside the sniffed prefix, and after it
+ f := &failingFile{data: make([]byte, 4096), good: good}
+ if _, _, err := c.tag(context.Background(), fmt.Sprintf("/nowhere/picture-%d", good), f, fi); err == nil {
+ t.Fatalf("hashing a file that fails after %d bytes reported no error", good)
+ }
+ if f.pos != 0 {
+ t.Fatalf("after a failed hash (%d good bytes) the file is at offset %d, want 0", good, f.pos)
+ }
+ }
+ if n := len(c.entries); n != 0 {
+ t.Fatalf("%d tags remembered from hashes that failed", n)
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index e7cdcb635..91300c1cc 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -805,7 +805,7 @@ func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps)
return
}
optDir := release.OptDir()
- restart := makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, bounceFn(svc))
+ restart := makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, adoptBounceFn(svc))
log.Printf("agent: adopt loop armed (hourly + jitter)")
// First pass shortly after boot (Eleven's OnBootSec=5min, compressed —
// the edge is already up by the time this fires), then hourly + jitter.
@@ -1376,6 +1376,7 @@ type fastDeps struct {
restartLeg func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
wakeLeg func(ctx context.Context) // the sidecar wake check (design: sidecar members), or a test hook
scheduleLeg func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
+ convergeLeg func() // finishes adoptions another process performed (convergeAdopted); nil = no activator armed
systemLeg *systemLeg // the platform self-update leg (spec §4); nil = not armed
lastSerial int64
seeded bool
@@ -1595,6 +1596,87 @@ func (d *fastDeps) observeRestart(restart func(string) error) func(string) error
}
}
+// convergeAdopted finishes the adoptions this process delegated or never saw.
+//
+// The agent is sandboxed (carlos-edge.service: ProtectSystem=strict, no /opt
+// write), so on a real box a new version is staged, pointed live and stamped
+// by a ROOT pass — carlos-stage.service on this loop's own request, or the
+// hourly carlos-adopt.service. That pass cannot cycle an exec child
+// (rootRestartRoute: the children are this process's), and once it has run,
+// this process's own pass finds the host up to date and never reaches the
+// restart callback RestartHibernating exists to hand it. The warm child was
+// left on the old build until the activator sweep's ConvergeGrace ran out:
+// 110-200s per deploy on the flagship, past `carlos deploy`'s own watch
+// (discussion #231).
+//
+// So, for each warm instance the activator reports drifted, cycle it once the
+// route is STAMPED with the version its symlink now names. The stamp is the
+// adopting pass saying it is finished with the host (adoptHost writes it last,
+// after its restart and any rollback); a drift the stamp does not yet agree
+// with is an adoption still in flight, and this leg waits a tick for it rather
+// than race it. execBounceable keeps the usual line between the two halves: a
+// route whose process the root pass's systemctl arms can reach is never this
+// leg's to cycle.
+//
+// At most limit instances are cycled per call, in host order. One pass can
+// adopt every warm instance of an app at once — a root pass repoints them all
+// inside a few seconds — and stopping them together would drain, checkpoint
+// and later restore them together too: every host of the app blips at the
+// same moment. The activator sweep spends the same care (one convergence
+// cycle per sweep); this leg keeps it at its own cadence, so N instances
+// converge over N ticks rather than in one. Only a cycle that actually
+// started counts against the limit — a drifted host this leg must wait on
+// costs the others nothing.
+//
+// Never wakes anything — ConvergeAdopted is a stop, and a sleeper execs the
+// new build at its own next wake. drifted returning nothing is the ordinary
+// tick and costs no registry read. Returns how many it cycled.
+func convergeAdopted(drifted func() map[string]string, get func(host string) (*registry.Route, error), converge func(host, ver string) bool, limit int) int {
+ stale := drifted()
+ if len(stale) == 0 {
+ return 0
+ }
+ hosts := make([]string, 0, len(stale))
+ for host := range stale {
+ hosts = append(hosts, host)
+ }
+ sort.Strings(hosts)
+ n := 0
+ for _, host := range hosts {
+ if n >= limit {
+ break
+ }
+ live := stale[host]
+ rt, err := get(host)
+ if err != nil || rt == nil || rt.Version != live || !execBounceable(*rt) {
+ continue
+ }
+ if converge(host, live) {
+ n++
+ log.Printf("agent: %s was adopted onto %s by another pass — cycled its warm instance onto it", host, live)
+ }
+ }
+ return n
+}
+
+// adoptedCyclesPerTick is convergeAdopted's limit on the fast loop: one warm
+// instance per 2s tick. A single-instance app — the common case — converges
+// on the first tick after its adoption is stamped; an app with a dozen warm
+// instances takes under half a minute, one blip at a time.
+const adoptedCyclesPerTick = 1
+
+// convergeLegFor builds the fast loop's converge leg over the box's
+// activator. nil for a box with none armed: there are no exec children to
+// cycle, and fastTick skips a nil leg.
+func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() {
+ if svc == nil {
+ return nil
+ }
+ return func() {
+ convergeAdopted(svc.Drifted, reg.Get, svc.ConvergeAdopted, adoptedCyclesPerTick)
+ }
+}
+
// fastTick is one fast-converge tick — two independent jobs (spec §3-§4):
//
// 1. Serial check: one ReadSerial. The FIRST successful read only seeds
@@ -1624,6 +1706,9 @@ func (d *fastDeps) observeRestart(restart func(string) error) func(string) error
// exactly those hosts — an empty sweep costs zero store reads. After
// the sweep, members still unstaged back off; staged members clear
// their backoff state.
+//
+// Between the two, every tick, the converge leg (convergeAdopted) cycles any
+// warm exec child that another process's adoption left on its old build.
func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
doc, _, err := release.ReadSerial(ctx, d.store)
switch {
@@ -1685,6 +1770,16 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
d.systemLeg.tick(ctx, now)
}
+ // The converge leg, on EVERY tick and before the sweep's early return. It
+ // cannot ride the serial switch above: the root pass that finishes a
+ // delegated adoption does so seconds AFTER the bump this loop acted on,
+ // and nothing bumps the serial again to mark it. Nor can it ride the
+ // sweep: an already-staged edge route is never a sweep member. It reads
+ // no store, and no registry row unless a warm child has actually drifted.
+ if d.convergeLeg != nil {
+ d.convergeLeg()
+ }
+
members := d.fastMembers(now)
if len(members) == 0 {
return // zero store reads: no PassFilter, no bake-clock write
@@ -1749,7 +1844,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
// sweep instead of the sweep re-cycling the service every tick. The
// remaining fields are filled in right below.
d := &fastDeps{reg: reg, store: store, pub: pub, optDir: optDir}
- restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, bounceFn(svc)))
+ restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, adoptBounceFn(svc)))
// Built once, used by both pass shapes below — see fastPassOpts for the
// DO-NOT list and why the stage requester is per-loop, not per-tick.
opts := fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit))
@@ -1782,6 +1877,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
}
d.restart = restart
d.passOpts = opts
+ d.convergeLeg = convergeLegFor(svc, reg)
// The serial-triggered full pass set, in spec order: mint routes
// (reconcile), then reconcile aliases, then stage them (adopt), then deliver
// their config. Each error is one log line, never fatal — the hourly floors
@@ -3049,6 +3145,19 @@ func bounceFn(svc *activator.Service) func(host string) bool {
return svc.Bounce
}
+// adoptBounceFn is bounceFn for an ADOPT pass's restart dispatch
+// (makeRestartRoute on adoptLoop and fastConvergeLoop): the same bounce,
+// except that a child already exec'd from the build live/<host> names is left
+// running — see activator.BounceAdopted for the interleaving that produces
+// one. `carlos restart` keeps bounceFn: a restart request cycles whatever is
+// running.
+func adoptBounceFn(svc *activator.Service) func(host string) bool {
+ if svc == nil {
+ return nil
+ }
+ return svc.BounceAdopted
+}
+
// wakeFn adapts an activator to cycleWorkers' wake parameter, nil-safe like
// bounceFn: a box with no activator armed gets a nil callback, which
// cycleWorkers reads as "nothing to wake" rather than panicking on a nil
diff --git a/internal/host/converge_adopted_test.go b/internal/host/converge_adopted_test.go
new file mode 100644
index 000000000..6934b1c88
--- /dev/null
+++ b/internal/host/converge_adopted_test.go
@@ -0,0 +1,416 @@
+package host
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "reflect"
+ "runtime"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The stamp is the whole licence for skipping the sweep's grace: it is written
+// only after the adopting pass is finished with the host. A drift whose route
+// is still stamped with the old version is an adoption in flight, and one with
+// no row at all is not a route this box adopts.
+func TestConvergeAdoptedCyclesOnlyWhenTheStampAgrees(t *testing.T) {
+ rows := map[string]*registry.Route{
+ "done.test": {Host: "done.test", Version: "v2"},
+ "inflight.test": {Host: "inflight.test", Version: "v1"},
+ "named.test": {Host: "named.test", Version: "v2", Unit: "console.service"},
+ "unitback.test": {Host: "unitback.test", Version: "v2", Backing: registry.BackingUnit},
+ }
+ drifted := func() map[string]string {
+ return map[string]string{
+ "done.test": "v2", "inflight.test": "v2", "named.test": "v2",
+ "unitback.test": "v2", "norow.test": "v2",
+ }
+ }
+ var cycled []string
+ n := convergeAdopted(drifted,
+ func(h string) (*registry.Route, error) { return rows[h], nil },
+ func(h, ver string) bool { cycled = append(cycled, h+"@"+ver); return true }, 10)
+ if want := []string{"done.test@v2"}; !reflect.DeepEqual(cycled, want) {
+ t.Fatalf("cycled = %v, want %v", cycled, want)
+ }
+ if n != 1 {
+ t.Fatalf("convergeAdopted = %d, want 1", n)
+ }
+}
+
+// The ordinary tick, where nothing has shipped, must cost no registry read:
+// this leg runs every two seconds on every box.
+func TestConvergeAdoptedReadsNoRouteWhenNothingDrifted(t *testing.T) {
+ n := convergeAdopted(func() map[string]string { return nil },
+ func(h string) (*registry.Route, error) {
+ t.Fatalf("read the route for %s with nothing drifted", h)
+ return nil, nil
+ },
+ func(h, ver string) bool {
+ t.Fatalf("cycled %s with nothing drifted", h)
+ return false
+ }, 10)
+ if n != 0 {
+ t.Fatalf("convergeAdopted = %d, want 0", n)
+ }
+}
+
+// The leg runs on EVERY tick, not only on a serial move: the root pass that
+// finishes the adoption does so seconds after the bump the agent acted on, and
+// nothing bumps the serial again to mark it. It also runs on a tick with no
+// sweep members, which is the tick an already-staged edge route always gets.
+func TestFastTickRunsTheConvergeLegEveryTick(t *testing.T) {
+ env := newFastFixture(t)
+ d := env.deps(nil)
+ var legs int
+ d.convergeLeg = func() { legs++ }
+ now := time.Unix(1754640000, 0)
+ ctx := context.Background()
+ fastTick(ctx, d, now)
+ fastTick(ctx, d, now.Add(2*time.Second))
+ if legs != 2 {
+ t.Fatalf("the converge leg ran %d times over two quiet ticks, want 2", legs)
+ }
+ d.convergeLeg = nil // a box with no activator armed
+ fastTick(ctx, d, now.Add(4*time.Second))
+}
+
+// The production sequence behind discussion #231, with the real pass on both
+// sides. The agent cannot write the version store, so the ROOT pass adopts a
+// new release for a warm exec instance: it stages, repoints and stamps, and —
+// having no handle on the agent's children — cycles nothing. The agent's own
+// pass then finds the host up to date and never reaches its restart callback.
+// Before the converge leg, that left the child on the old build until the
+// activator sweep's 90s grace ran out.
+func TestRootAdoptionOfAWarmExecInstanceIsFinishedByTheAgent(t *testing.T) {
+ env := newFastFixture(t)
+ ctx := context.Background()
+ plat := runtime.GOOS + "-" + runtime.GOARCH
+ host := "warm.test"
+ ship := func(ver string) {
+ t.Helper()
+ m := release.Manifest{
+ App: "fastapp", Version: ver, Kind: "binary",
+ Artifacts: map[string]release.Artifact{plat: {}},
+ }
+ if err := release.Ship(ctx, env.store, "bab", m, map[string][]byte{plat: []byte("build " + ver)}); err != nil {
+ t.Fatalf("Ship %s: %v", ver, err)
+ }
+ if err := release.Promote(ctx, env.store, env.priv, "bab", "fastapp", "edge", ver); err != nil {
+ t.Fatalf("Promote %s: %v", ver, err)
+ }
+ }
+ // rootRestartRoute with systemctl stubbed out: same dispatch, bounce nil.
+ rootPass := func() {
+ t.Helper()
+ restart := makeRestartRoute(env.reg, func(string) {}, func(string) error { return nil }, nil)
+ if err := release.Pass(ctx, env.reg, env.store, env.pub, env.optDir, restart, release.BakeDir(env.dataDir)); err != nil {
+ t.Fatalf("root pass: %v", err)
+ }
+ }
+ if err := env.reg.Upsert(registry.Route{
+ Host: host, Kind: "instance", App: "fastapp", Channel: "edge",
+ Status: "ready", Hibernate: true,
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ ship("v1")
+ rootPass()
+ spawned := "v1" // the awake child's build: frozen at exec, as the activator records it
+
+ ship("v2")
+ rootPass()
+ if v, err := release.LiveVersion(env.optDir, host); err != nil || v != "v2" {
+ t.Fatalf("live = %q, %v after the root pass; want v2", v, err)
+ }
+ if rt, err := env.reg.Get(host); err != nil || rt == nil || rt.Version != "v2" {
+ t.Fatalf("route = %+v, %v after the root pass; want it stamped v2", rt, err)
+ }
+
+ // The agent's tick. Its pass has a bounce to offer and never gets to use
+ // it: the host is already up to date.
+ var bounced, cycled []string
+ restart := makeRestartRoute(env.reg, func(string) {}, func(string) error { return nil },
+ func(h string) bool { bounced = append(bounced, h); return true })
+ d := env.deps(nil)
+ d.restart = restart
+ d.passOpts = fastPassOpts(env.dataDir, func(string, string, error) {}, restart, nil)
+ d.fullPass = func(ctx context.Context) {
+ if err := release.Pass(ctx, d.reg, d.store, d.pub, d.optDir, d.restart, d.passOpts...); err != nil {
+ t.Fatalf("agent pass: %v", err)
+ }
+ }
+ d.convergeLeg = func() {
+ convergeAdopted(
+ func() map[string]string {
+ live, _ := release.LiveVersion(env.optDir, host)
+ if live == "" || live == spawned {
+ return nil
+ }
+ return map[string]string{host: live}
+ },
+ env.reg.Get,
+ func(h, ver string) bool { cycled = append(cycled, h+"@"+ver); spawned = ver; return true },
+ adoptedCyclesPerTick)
+ }
+ now := time.Unix(1754640000, 0)
+ fastTick(ctx, d, now) // seeds the serial
+ if err := release.BumpSerial(ctx, env.store, now); err != nil {
+ t.Fatalf("BumpSerial: %v", err)
+ }
+ fastTick(ctx, d, now.Add(10*time.Second)) // the serial move: full pass, then the leg
+
+ if len(bounced) != 0 {
+ t.Fatalf("the agent's pass bounced %v — it found the host already adopted, so this test no longer models the gap", bounced)
+ }
+ if want := []string{host + "@v2"}; !reflect.DeepEqual(cycled, want) {
+ t.Fatalf("cycled = %v, want %v: the warm child is still on v1 under a symlink and a stamp that both say v2", cycled, want)
+ }
+ fastTick(ctx, d, now.Add(12*time.Second))
+ if len(cycled) != 1 {
+ t.Fatalf("cycled again on the next tick: %v", cycled)
+ }
+}
+
+// One pass can adopt every warm instance of an app inside a few seconds.
+// Stopping them all on one tick would blip every host of the app at the same
+// moment; the leg spends its budget in host order and leaves the rest for the
+// ticks that follow. A drifted host it has to wait on costs the others nothing.
+func TestConvergeAdoptedCyclesOnePerTickInHostOrder(t *testing.T) {
+ spawned := map[string]string{"a.test": "v1", "b.test": "v1", "c.test": "v1", "wait.test": "v1"}
+ stamp := map[string]string{"a.test": "v2", "b.test": "v2", "c.test": "v2", "wait.test": "v1"}
+ drifted := func() map[string]string {
+ out := map[string]string{}
+ for h, v := range spawned {
+ if v != "v2" {
+ out[h] = "v2"
+ }
+ }
+ return out
+ }
+ get := func(h string) (*registry.Route, error) { return &registry.Route{Host: h, Version: stamp[h]}, nil }
+ var cycled []string
+ converge := func(h, ver string) bool { cycled = append(cycled, h); spawned[h] = ver; return true }
+
+ for tick := 1; tick <= 4; tick++ {
+ if n := convergeAdopted(drifted, get, converge, adoptedCyclesPerTick); tick <= 3 && n != 1 {
+ t.Fatalf("tick %d cycled %d, want 1", tick, n)
+ }
+ }
+ if want := []string{"a.test", "b.test", "c.test"}; !reflect.DeepEqual(cycled, want) {
+ t.Fatalf("cycled = %v, want %v: one per tick, in host order, never the host whose stamp does not agree", cycled, want)
+ }
+}
+
+// versionedBackend is activator.Backend with the one thing convergence needs
+// and fakeWakeBackend does not model: a child that knows which build it was
+// exec'd from, read off the SAME live symlink the release pass writes. It
+// binds no socket — the caller supplies the proxy — so it runs anywhere.
+type versionedBackend struct {
+ fakeWakeBackend
+ optDir string
+
+ mu sync.Mutex
+ starts []string // the build each spawn exec'd, in order
+}
+
+type flagProc struct {
+ mu sync.Mutex
+ stopped bool
+}
+
+func (p *flagProc) Stop(context.Context) error {
+ p.mu.Lock()
+ p.stopped = true
+ p.mu.Unlock()
+ return nil
+}
+
+func (p *flagProc) Exited() bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.stopped
+}
+
+func (b *versionedBackend) LiveVersion(host string) (string, error) {
+ return release.LiveVersion(b.optDir, host)
+}
+
+func (b *versionedBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) {
+ ver, err := release.LiveVersion(b.optDir, rt.Host)
+ if err != nil {
+ return nil, "", err
+ }
+ b.mu.Lock()
+ b.starts = append(b.starts, ver)
+ b.mu.Unlock()
+ return &flagProc{}, ver, nil
+}
+
+func (b *versionedBackend) started() []string {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return append([]string(nil), b.starts...)
+}
+
+// The whole of discussion #231 with nothing stood in for: the real release
+// pass as root runs it, a real activator owning the child, and the leg exactly
+// as fastConvergeLoop builds it (convergeLegFor). The sweeper is never
+// started, so the ONLY thing that can cycle the instance is the leg — a 90s
+// grace cannot rescue this test.
+func TestConvergeLegCyclesARootAdoptedInstanceOnARealActivator(t *testing.T) {
+ env := newFastFixture(t)
+ ctx := context.Background()
+ plat := runtime.GOOS + "-" + runtime.GOARCH
+ host := "warm.test"
+ rt := registry.Route{
+ Host: host, Kind: "instance", App: "fastapp", Channel: "edge",
+ Status: "ready", Hibernate: true,
+ }
+ if err := env.reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ adopt := func(ver string) {
+ t.Helper()
+ m := release.Manifest{
+ App: "fastapp", Version: ver, Kind: "binary",
+ Artifacts: map[string]release.Artifact{plat: {}},
+ }
+ if err := release.Ship(ctx, env.store, "bab", m, map[string][]byte{plat: []byte("build " + ver)}); err != nil {
+ t.Fatalf("Ship %s: %v", ver, err)
+ }
+ if err := release.Promote(ctx, env.store, env.priv, "bab", "fastapp", "edge", ver); err != nil {
+ t.Fatalf("Promote %s: %v", ver, err)
+ }
+ // The root pass: rootRestartRoute's dispatch (bounce nil), systemctl stubbed.
+ restart := makeRestartRoute(env.reg, func(string) {}, func(string) error { return nil }, nil)
+ if err := release.Pass(ctx, env.reg, env.store, env.pub, env.optDir, restart, release.BakeDir(env.dataDir)); err != nil {
+ t.Fatalf("root pass for %s: %v", ver, err)
+ }
+ }
+ backend := &versionedBackend{optDir: env.optDir}
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, backend, env.reg)
+ t.Cleanup(func() {
+ sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(sctx)
+ })
+ request := func() {
+ t.Helper()
+ rec := httptest.NewRecorder()
+ proxy := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
+ if !svc.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+host+"/", nil), rt, proxy) {
+ t.Fatal("the activator did not take the request")
+ }
+ if rec.Code != http.StatusNoContent {
+ t.Fatalf("request = %d, want 204", rec.Code)
+ }
+ }
+ leg := convergeLegFor(svc, env.reg)
+ if leg == nil {
+ t.Fatal("convergeLegFor returned no leg for a box with an activator")
+ }
+
+ // "Was it stopped?" is asked of the spawn count, after a request: Owns
+ // stays true while an entry drains, so it cannot tell a spared child from
+ // one being stopped. A request against a stopped instance spawns again.
+ adopt("v1")
+ request() // wakes the instance onto v1
+ leg()
+ request()
+ if got, want := backend.started(), []string{"v1"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the leg stopped an instance whose build is the adopted one", got, want)
+ }
+
+ adopt("v2") // root: staged, repointed, stamped — and nothing cycled
+ if !svc.Owns(host) {
+ t.Fatal("the root pass stopped the agent's child — this test no longer models the gap")
+ }
+ leg()
+ deadline := time.Now().Add(10 * time.Second)
+ for svc.Owns(host) {
+ if time.Now().After(deadline) {
+ t.Fatal("the warm instance is still running v1 after the leg: it was adopted onto v2 and nothing cycled it")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+
+ request() // the next request wakes it onto v2
+ if got, want := backend.started(), []string{"v1", "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v", got, want)
+ }
+ leg()
+ request()
+ if got, want := backend.started(), []string{"v1", "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the leg cycled the instance again after it had converged", got, want)
+ }
+}
+
+func TestConvergeLegForABoxWithNoActivator(t *testing.T) {
+ if convergeLegFor(nil, nil) != nil {
+ t.Fatal("convergeLegFor built a leg with no activator to ask")
+ }
+}
+
+// The adopt passes' restart dispatch must be built on adoptBounceFn, not
+// bounceFn. With the converge leg in place an instance can already be on the
+// new build when a slower pass reaches its own restart; the plain Bounce would
+// stop that replacement for nothing. A stale child still gets its cycle.
+func TestAdoptRestartDispatchSparesAConvergedChild(t *testing.T) {
+ env := newFastFixture(t)
+ host := "warm.test"
+ rt := registry.Route{
+ Host: host, Kind: "instance", App: "fastapp", Channel: "edge",
+ Status: "ready", Hibernate: true,
+ }
+ if err := env.reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ env.stageLive(t, host, "v1")
+ backend := &versionedBackend{optDir: env.optDir}
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, backend, env.reg)
+ t.Cleanup(func() {
+ sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(sctx)
+ })
+ wake := func() {
+ t.Helper()
+ rec := httptest.NewRecorder()
+ proxy := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
+ if !svc.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+host+"/", nil), rt, proxy) || rec.Code != http.StatusNoContent {
+ t.Fatalf("wake request = %d", rec.Code)
+ }
+ }
+ restart := makeRestartRoute(env.reg, func(string) {}, func(u string) error {
+ t.Fatalf("named-unit arm fired for %q on a route that names no unit", u)
+ return nil
+ }, adoptBounceFn(svc))
+
+ // Asked of the spawn count, after a request — see the note in
+ // TestConvergeLegCyclesARootAdoptedInstanceOnARealActivator.
+ wake() // exec'd from v1, and live/<host> names v1
+ if err := restart(host); err != nil {
+ t.Fatalf("restart: %v", err)
+ }
+ wake()
+ if got, want := backend.started(), []string{"v1"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the adopt dispatch stopped a child already running the build live/<host> names", got, want)
+ }
+
+ env.stageLive(t, host, "v2") // the symlink moves under the running child
+ if err := restart(host); err != nil {
+ t.Fatalf("restart: %v", err)
+ }
+ wake()
+ if got, want := backend.started(), []string{"v1", "v2"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("spawned builds = %v, want %v: the adopt dispatch left a stale child running", got, want)
+ }
+}
diff --git a/internal/lease/export_test.go b/internal/lease/export_test.go
new file mode 100644
index 000000000..c6c589e69
--- /dev/null
+++ b/internal/lease/export_test.go
@@ -0,0 +1,9 @@
+package lease
+
+import awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+
+// NewVersionedS3WithClients is the test-only constructor: tests bind
+// buckets to clients pointed at leasetest.NewS3Server.
+func NewVersionedS3WithClients(clients map[string]*awss3.Client) VersionedStore {
+ return &versionedS3{clients: clients}
+}
diff --git a/internal/lease/leasetest/bucket.go b/internal/lease/leasetest/bucket.go
new file mode 100644
index 000000000..fe13f4d98
--- /dev/null
+++ b/internal/lease/leasetest/bucket.go
@@ -0,0 +1,340 @@
+// Package leasetest holds test doubles for internal/lease. VersionedBucket
+// is the reference model of a versioned S3 bucket that every plan of the
+// host lifecycle claim tests against (spec §13).
+package leasetest
+
+import (
+ "bytes"
+ "context"
+ "crypto/md5"
+ "encoding/hex"
+ "fmt"
+ "io"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+type version struct {
+ id string
+ body []byte
+ etag string
+ deleteMarker bool
+ modified time.Time
+}
+
+// VersionedBucket is an in-memory, versioned, multi-bucket S3 model.
+type VersionedBucket struct {
+ mu sync.Mutex
+ buckets map[string]map[string][]version // bucket -> key -> versions, newest first
+ next int
+ clock time.Time
+
+ PageSize int // listing page size; default 1000
+ FailListAfterPages int // fail a listing once this many pages have been served; 0 never
+ FailOn func(op, key string) error // op: list|get|copy|delete|put
+}
+
+func NewVersionedBucket() *VersionedBucket {
+ return &VersionedBucket{buckets: map[string]map[string][]version{}, PageSize: 1000,
+ clock: time.Date(2026, 10, 5, 0, 0, 0, 0, time.UTC)}
+}
+
+func md5hex(b []byte) string { s := md5.Sum(b); return hex.EncodeToString(s[:]) }
+
+func (f *VersionedBucket) AddBucket(name string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if f.buckets[name] == nil {
+ f.buckets[name] = map[string][]version{}
+ }
+}
+
+// --- locked helpers: callers hold f.mu ---
+
+func (f *VersionedBucket) fail(op, key string) error {
+ if f.FailOn != nil {
+ return f.FailOn(op, key)
+ }
+ return nil
+}
+
+func (f *VersionedBucket) known(bucket string) error {
+ if _, ok := f.buckets[bucket]; !ok {
+ return fmt.Errorf("leasetest: no bucket %q", bucket)
+ }
+ return nil
+}
+
+func (f *VersionedBucket) push(bucket, key string, v version) string {
+ f.next++
+ f.clock = f.clock.Add(time.Second)
+ v.id, v.modified = fmt.Sprintf("v%06d", f.next), f.clock
+ f.buckets[bucket][key] = append([]version{v}, f.buckets[bucket][key]...)
+ return v.id
+}
+
+func (f *VersionedBucket) hasCurrent(bucket, key string) bool {
+ vs := f.buckets[bucket][key]
+ return len(vs) > 0 && !vs[0].deleteMarker
+}
+
+func (f *VersionedBucket) find(bucket, key, id string) (version, bool) {
+ for _, v := range f.buckets[bucket][key] {
+ if v.id == id {
+ return v, true
+ }
+ }
+ return version{}, false
+}
+
+// listAll is the whole ordered listing under prefix (key order, newest
+// first within a key).
+func (f *VersionedBucket) listAll(bucket, prefix string) []lease.ObjectVersion {
+ o := f.buckets[bucket]
+ keys := make([]string, 0, len(o))
+ for k := range o {
+ if strings.HasPrefix(k, prefix) {
+ keys = append(keys, k)
+ }
+ }
+ sort.Strings(keys)
+ var all []lease.ObjectVersion
+ for _, k := range keys {
+ for i, v := range o[k] {
+ ov := lease.ObjectVersion{Key: k, VersionID: v.id, IsLatest: i == 0, DeleteMarker: v.deleteMarker, LastModified: v.modified}
+ if !v.deleteMarker {
+ ov.ETag, ov.Size = v.etag, int64(len(v.body))
+ }
+ all = append(all, ov)
+ }
+ }
+ return all
+}
+
+// pageOf returns page number page (0-based) of all, whether more follow,
+// and the failure FailListAfterPages injects.
+func (f *VersionedBucket) pageOf(all []lease.ObjectVersion, page int) ([]lease.ObjectVersion, bool, error) {
+ if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
+ return nil, false, fmt.Errorf("leasetest: injected listing failure on page %d", page+1)
+ }
+ size := f.PageSize
+ if size <= 0 {
+ size = 1000
+ }
+ start := page * size
+ if start > len(all) {
+ start = len(all)
+ }
+ end := start + size
+ if end >= len(all) {
+ return all[start:], false, nil
+ }
+ return all[start:end], true, nil
+}
+
+// --- exported helpers for tests ---
+
+func (f *VersionedBucket) Put(bucket, key string, body []byte) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)})
+}
+
+func (f *VersionedBucket) PutMultipart(bucket, key string, body []byte) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body) + "-2"})
+}
+
+func (f *VersionedBucket) BareDelete(bucket, key string) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.push(bucket, key, version{deleteMarker: true})
+}
+
+func (f *VersionedBucket) LatestID(bucket, key string) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if vs := f.buckets[bucket][key]; len(vs) > 0 {
+ return vs[0].id
+ }
+ return ""
+}
+
+func (f *VersionedBucket) Current(bucket, key string) ([]byte, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ vs := f.buckets[bucket][key]
+ if len(vs) == 0 || vs[0].deleteMarker {
+ return nil, false
+ }
+ return vs[0].body, true
+}
+
+// Snapshot renders the whole state, for before/after comparison.
+func (f *VersionedBucket) Snapshot() string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ var b strings.Builder
+ names := make([]string, 0, len(f.buckets))
+ for n := range f.buckets {
+ names = append(names, n)
+ }
+ sort.Strings(names)
+ for _, n := range names {
+ for _, v := range f.listAll(n, "") {
+ fmt.Fprintf(&b, "%s/%s@%s:%v:%s\n", n, v.Key, v.VersionID, v.DeleteMarker, v.ETag)
+ }
+ }
+ return b.String()
+}
+
+// --- lease.VersionedStore ---
+
+func (f *VersionedBucket) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if err := f.known(bucket); err != nil {
+ return nil, err
+ }
+ if err := f.fail("list", prefix); err != nil {
+ return nil, err
+ }
+ all := f.listAll(bucket, prefix)
+ var out []lease.ObjectVersion
+ for page := 0; ; page++ {
+ vs, more, err := f.pageOf(all, page)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, vs...)
+ if !more {
+ return out, nil
+ }
+ }
+}
+
+func (f *VersionedBucket) read(bucket, key, versionID string) ([]byte, error) {
+ if versionID == "" {
+ return nil, lease.ErrNoVersionID
+ }
+ if err := f.known(bucket); err != nil {
+ return nil, err
+ }
+ if err := f.fail("get", key); err != nil {
+ return nil, err
+ }
+ v, ok := f.find(bucket, key, versionID)
+ if !ok {
+ return nil, lease.ErrMirrorNotFound
+ }
+ if v.deleteMarker {
+ return nil, lease.ErrIsDeleteMarker
+ }
+ return append([]byte(nil), v.body...), nil
+}
+
+func (f *VersionedBucket) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.read(bucket, key, versionID)
+}
+
+func (f *VersionedBucket) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ b, err := f.read(bucket, key, versionID)
+ if err != nil {
+ return nil, err
+ }
+ return io.NopCloser(bytes.NewReader(b)), nil
+}
+
+func (f *VersionedBucket) copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ if srcVersionID == "" {
+ return "", "", lease.ErrNoVersionID
+ }
+ if err := f.known(srcBucket); err != nil {
+ return "", "", err
+ }
+ if err := f.known(dstBucket); err != nil {
+ return "", "", err
+ }
+ if err := f.fail("copy", srcKey); err != nil {
+ return "", "", err
+ }
+ v, ok := f.find(srcBucket, srcKey, srcVersionID)
+ if !ok {
+ return "", "", lease.ErrMirrorNotFound
+ }
+ if v.deleteMarker {
+ return "", "", lease.ErrIsDeleteMarker
+ }
+ if createOnly && f.hasCurrent(dstBucket, dstKey) {
+ return "", "", lease.ErrDestinationExists
+ }
+ etag := md5hex(v.body) // a copy is single-part (§13 row 9)
+ return f.push(dstBucket, dstKey, version{body: append([]byte(nil), v.body...), etag: etag}), etag, nil
+}
+
+func (f *VersionedBucket) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly)
+}
+
+func (f *VersionedBucket) deleteLocked(bucket, key, versionID string) error {
+ if versionID == "" {
+ return lease.ErrNoVersionID
+ }
+ if err := f.known(bucket); err != nil {
+ return err
+ }
+ if err := f.fail("delete", key); err != nil {
+ return err
+ }
+ vs := f.buckets[bucket][key]
+ for i, v := range vs {
+ if v.id == versionID {
+ rest := append(vs[:i:i], vs[i+1:]...)
+ if len(rest) == 0 {
+ delete(f.buckets[bucket], key)
+ } else {
+ f.buckets[bucket][key] = rest
+ }
+ return nil
+ }
+ }
+ return nil
+}
+
+func (f *VersionedBucket) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.deleteLocked(bucket, key, versionID)
+}
+
+func (f *VersionedBucket) putIfAbsentLocked(bucket, key string, body []byte) (string, error) {
+ if err := f.known(bucket); err != nil {
+ return "", err
+ }
+ if err := f.fail("put", key); err != nil {
+ return "", err
+ }
+ if f.hasCurrent(bucket, key) {
+ return "", lease.ErrExists
+ }
+ return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil
+}
+
+func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.putIfAbsentLocked(bucket, key, body)
+}
+
+var _ lease.VersionedStore = (*VersionedBucket)(nil)
diff --git a/internal/lease/leasetest/bucket_test.go b/internal/lease/leasetest/bucket_test.go
new file mode 100644
index 000000000..d711a6f49
--- /dev/null
+++ b/internal/lease/leasetest/bucket_test.go
@@ -0,0 +1,15 @@
+package leasetest
+
+import (
+ "testing"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+func TestVersionedBucketContract(t *testing.T) {
+ RunVersionedStoreContract(t, func(t *testing.T) (lease.VersionedStore, *VersionedBucket, string, func() int) {
+ m := NewVersionedBucket()
+ m.AddBucket("data")
+ return m, m, "data", nil
+ })
+}
diff --git a/internal/lease/leasetest/contract.go b/internal/lease/leasetest/contract.go
new file mode 100644
index 000000000..d1a823408
--- /dev/null
+++ b/internal/lease/leasetest/contract.go
@@ -0,0 +1,245 @@
+package leasetest
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+// Maker builds a store under test, the model backing it, the bucket name,
+// and (for wire stores) a request counter.
+type Maker func(t *testing.T) (st lease.VersionedStore, m *VersionedBucket, bucket string, served func() int)
+
+// RunVersionedStoreContract is the behaviour every lease.VersionedStore
+// must have (spec §13, as covered by Plan 2).
+func RunVersionedStoreContract(t *testing.T, mk Maker) {
+ ctx := context.Background()
+
+ t.Run("listing: key order, latest first, markers included, prefix bounded", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ m.Put(b, "p/a", []byte("one"))
+ m.Put(b, "p/a", []byte("two"))
+ m.BareDelete(b, "p/a")
+ m.Put(b, "p/b", []byte("x"))
+ m.Put(b, "q/c", []byte("other prefix"))
+ vs, err := st.ListVersions(ctx, b, "p/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(vs) != 4 || vs[0].Key != "p/a" || !vs[0].IsLatest || !vs[0].DeleteMarker || vs[3].Key != "p/b" || !vs[3].IsLatest {
+ t.Fatalf("listing = %+v", vs)
+ }
+ for _, v := range vs[1:3] {
+ if v.Key != "p/a" || v.IsLatest || v.DeleteMarker || v.VersionID == "" {
+ t.Fatalf("non-latest entries = %+v", vs[1:3])
+ }
+ }
+ })
+
+ t.Run("listing spans pages, including many versions of one key", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ m.PageSize = 3
+ for i := 0; i < 5; i++ {
+ m.Put(b, "p/same", []byte{byte(i)})
+ }
+ for i := 0; i < 4; i++ {
+ m.Put(b, fmt.Sprintf("p/k%02d", i), []byte{byte(i)})
+ }
+ vs, err := st.ListVersions(ctx, b, "p/")
+ if err != nil || len(vs) != 9 {
+ t.Fatalf("paged listing = %d, %v; want 9", len(vs), err)
+ }
+ seen := map[string]bool{}
+ for _, v := range vs {
+ if seen[v.Key+"@"+v.VersionID] {
+ t.Fatalf("duplicate %s@%s", v.Key, v.VersionID)
+ }
+ seen[v.Key+"@"+v.VersionID] = true
+ }
+ })
+
+ t.Run("listing fails whole on a page error", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ m.PageSize = 2
+ for i := 0; i < 6; i++ {
+ m.Put(b, fmt.Sprintf("p/k%d", i), []byte{byte(i)})
+ }
+ m.FailListAfterPages = 2
+ if vs, err := st.ListVersions(ctx, b, "p/"); err == nil || vs != nil {
+ t.Fatalf("got %d entries and error %v; want no entries and an error", len(vs), err)
+ }
+ })
+
+ t.Run("read and stream by exact version; markers and missing", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ k := "dir/a b+c%?d"
+ v1 := m.Put(b, k, []byte("old"))
+ m.Put(b, k, []byte("new"))
+ if body, err := st.GetVersion(ctx, b, k, v1); err != nil || string(body) != "old" {
+ t.Fatalf("GetVersion = %q, %v", body, err)
+ }
+ rc, err := st.OpenVersion(ctx, b, k, v1)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, _ := io.ReadAll(rc)
+ rc.Close()
+ if string(body) != "old" {
+ t.Fatalf("OpenVersion = %q", body)
+ }
+ mk1 := m.BareDelete(b, k)
+ if _, err := st.GetVersion(ctx, b, k, mk1); !errors.Is(err, lease.ErrIsDeleteMarker) {
+ t.Fatalf("GetVersion(marker) = %v, want ErrIsDeleteMarker", err)
+ }
+ if _, err := st.GetVersion(ctx, b, k, "v999999"); !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("GetVersion(missing) = %v, want ErrMirrorNotFound", err)
+ }
+ // A version hidden behind a marker is still readable by id.
+ if body, err := st.GetVersion(ctx, b, k, v1); err != nil || string(body) != "old" {
+ t.Fatalf("GetVersion behind a marker = %q, %v", body, err)
+ }
+ })
+
+ t.Run("copy by exact version; create-only; markers", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ k := "src/a b+c%?d"
+ v1 := m.Put(b, k, []byte("old"))
+ m.Put(b, k, []byte("new"))
+ dv, etag, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true)
+ if err != nil || dv == "" || etag == "" || strings.Contains(etag, "-") {
+ t.Fatalf("CopyVersion = %q %q %v", dv, etag, err)
+ }
+ if cur, ok := m.Current(b, "dst/x"); !ok || string(cur) != "old" {
+ t.Fatalf("copied %q", cur)
+ }
+ if _, _, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true); !errors.Is(err, lease.ErrDestinationExists) {
+ t.Fatalf("create-only copy onto existing = %v", err)
+ }
+ m.BareDelete(b, "dst/x")
+ if _, _, err := st.CopyVersion(ctx, b, k, v1, b, "dst/x", true); err != nil {
+ t.Fatalf("create-only copy over a delete marker = %v", err)
+ }
+ mid := m.BareDelete(b, k)
+ if _, _, err := st.CopyVersion(ctx, b, k, mid, b, "dst/y", false); !errors.Is(err, lease.ErrIsDeleteMarker) {
+ t.Fatalf("copy of a marker = %v, want ErrIsDeleteMarker", err)
+ }
+ mpID := m.PutMultipart(b, "src/mp", []byte("payload"))
+ if _, etag, err := st.CopyVersion(ctx, b, "src/mp", mpID, b, "dst/mp", false); err != nil || strings.Contains(etag, "-") {
+ t.Fatalf("multipart copy etag = %q, %v; want single-part", etag, err)
+ }
+ })
+
+ t.Run("delete by version: exact, idempotent, never bare", func(t *testing.T) {
+ st, m, b, served := mk(t)
+ v1 := m.Put(b, "k", []byte("one"))
+ v2 := m.Put(b, "k", []byte("two"))
+ if err := st.DeleteVersion(ctx, b, "k", v2); err != nil {
+ t.Fatal(err)
+ }
+ if cur, ok := m.Current(b, "k"); !ok || string(cur) != "one" {
+ t.Fatalf("after deleting the current version: %q %v; want the older one current", cur, ok)
+ }
+ if err := st.DeleteVersion(ctx, b, "k", v2); err != nil {
+ t.Fatalf("re-delete = %v", err)
+ }
+ if err := st.DeleteVersion(ctx, b, "never", "v000001"); err != nil {
+ t.Fatalf("delete on a missing key = %v", err)
+ }
+ before, reqs := m.Snapshot(), 0
+ if served != nil {
+ reqs = served()
+ }
+ if err := st.DeleteVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("empty version id = %v, want ErrNoVersionID", err)
+ }
+ if _, err := st.GetVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("GetVersion(\"\") = %v, want ErrNoVersionID", err)
+ }
+ if _, err := st.OpenVersion(ctx, b, "k", ""); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("OpenVersion(\"\") = %v, want ErrNoVersionID", err)
+ }
+ if _, _, err := st.CopyVersion(ctx, b, "k", "", b, "k2", false); !errors.Is(err, lease.ErrNoVersionID) {
+ t.Fatalf("CopyVersion(\"\") = %v, want ErrNoVersionID", err)
+ }
+ if m.Snapshot() != before {
+ t.Fatal("state changed on an empty-version delete")
+ }
+ if served != nil && served() != reqs {
+ t.Fatal("a request was sent for an empty version id")
+ }
+ _ = v1
+ })
+
+ t.Run("delete a noncurrent version keeps the current one", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ v1 := m.Put(b, "k", []byte("one"))
+ v2 := m.Put(b, "k", []byte("two"))
+ if err := st.DeleteVersion(ctx, b, "k", v1); err != nil {
+ t.Fatal(err)
+ }
+ if cur, ok := m.Current(b, "k"); !ok || string(cur) != "two" || m.LatestID(b, "k") != v2 {
+ t.Fatalf("current after a noncurrent delete = %q %v %s", cur, ok, m.LatestID(b, "k"))
+ }
+ if _, err := st.GetVersion(ctx, b, "k", v1); !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("deleted version still readable: %v", err)
+ }
+ })
+
+ t.Run("a missing bucket is an error, never absence", func(t *testing.T) {
+ st, _, _, _ := mk(t)
+ if err := st.DeleteVersion(ctx, "nobucket", "k", "v000001"); err == nil {
+ t.Error("delete in a missing bucket succeeded")
+ }
+ if _, err := st.GetVersion(ctx, "nobucket", "k", "v000001"); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Errorf("get in a missing bucket = %v", err)
+ }
+ if _, err := st.ListVersions(ctx, "nobucket", ""); err == nil {
+ t.Error("list of a missing bucket succeeded")
+ }
+ })
+
+ t.Run("put if absent", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ if v, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); err != nil || v == "" {
+ t.Fatalf("first = %q %v", v, err)
+ }
+ if _, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); !errors.Is(err, lease.ErrExists) {
+ t.Fatalf("second = %v, want ErrExists", err)
+ }
+ m.BareDelete(b, "m/closed.json")
+ if _, err := st.PutIfAbsent(ctx, b, "m/closed.json", []byte("{}")); err != nil {
+ t.Fatalf("over a marker = %v", err)
+ }
+ })
+
+ t.Run("injected failures surface as errors from every operation", func(t *testing.T) {
+ st, m, b, _ := mk(t)
+ v := m.Put(b, "k", []byte("x"))
+ boom := errors.New("boom")
+ m.FailOn = func(op, key string) error { return boom }
+ if _, err := st.ListVersions(ctx, b, ""); err == nil {
+ t.Error("list: no error")
+ }
+ if _, err := st.GetVersion(ctx, b, "k", v); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Errorf("get: %v", err)
+ }
+ if _, _, err := st.CopyVersion(ctx, b, "k", v, b, "k2", false); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Errorf("copy: %v", err)
+ }
+ if err := st.DeleteVersion(ctx, b, "k", v); err == nil {
+ t.Error("delete: no error")
+ }
+ if _, err := st.PutIfAbsent(ctx, b, "new", []byte("x")); err == nil {
+ t.Error("put: no error")
+ }
+ m.FailOn = nil
+ if _, ok := m.Current(b, "k"); !ok {
+ t.Fatal("a failed delete removed data")
+ }
+ })
+}
diff --git a/internal/lease/leasetest/s3server.go b/internal/lease/leasetest/s3server.go
new file mode 100644
index 000000000..f0877aeb5
--- /dev/null
+++ b/internal/lease/leasetest/s3server.go
@@ -0,0 +1,222 @@
+package leasetest
+
+import (
+ "encoding/xml"
+ "errors"
+ "io"
+ "net/http"
+ "net/url"
+ "strings"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+type xmlVersion struct {
+ XMLName xml.Name `xml:"Version"`
+ Key string
+ VersionId string
+ IsLatest bool
+ LastModified string
+ ETag string
+ Size int64
+}
+
+type xmlMarker struct {
+ XMLName xml.Name `xml:"DeleteMarker"`
+ Key string
+ VersionId string
+ IsLatest bool
+ LastModified string
+}
+
+type listVersionsResult struct {
+ XMLName xml.Name `xml:"ListVersionsResult"`
+ Name string
+ Prefix string
+ KeyMarker string
+ VersionIdMarker string
+ NextKeyMarker string `xml:",omitempty"`
+ NextVersionIdMarker string `xml:",omitempty"`
+ MaxKeys int
+ IsTruncated bool
+ Entries []any
+}
+
+func s3err(w http.ResponseWriter, status int, code string) { s3errMsg(w, status, code, code) }
+
+func s3errMsg(w http.ResponseWriter, status int, code, msg string) {
+ w.Header().Set("Content-Type", "application/xml")
+ w.WriteHeader(status)
+ io.WriteString(w, "<Error><Code>"+code+"</Code><Message>"+msg+"</Message></Error>")
+}
+
+// copyMarkerMsg is S3's answer to a COPY naming a delete marker's version.
+const copyMarkerMsg = "The source of a copy request may not specifically refer to a delete marker by version id."
+
+// writeErr maps a model error onto the S3 response real S3 gives.
+func writeErr(w http.ResponseWriter, err error) {
+ switch {
+ case errors.Is(err, lease.ErrMirrorNotFound):
+ s3err(w, http.StatusNotFound, "NoSuchVersion")
+ case errors.Is(err, lease.ErrIsDeleteMarker):
+ w.Header().Set("x-amz-delete-marker", "true")
+ s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed")
+ case errors.Is(err, lease.ErrDestinationExists), errors.Is(err, lease.ErrExists):
+ s3err(w, http.StatusPreconditionFailed, "PreconditionFailed")
+ default:
+ s3err(w, http.StatusInternalServerError, "InternalError")
+ }
+}
+
+// NewS3Server speaks the subset of the S3 wire protocol VersionedStore uses,
+// path-style, over f, holding f.mu for each request. The second result
+// counts requests served. With serves non-empty, the endpoint answers only
+// for those request buckets (a regional endpoint); any other bucket gets
+// 301 PermanentRedirect, as S3 answers a wrong-region request. A copy's
+// source bucket may be any bucket in f.
+func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int) {
+ var n atomic.Int64
+ allowed := map[string]bool{}
+ for _, b := range serves {
+ allowed[b] = true
+ }
+ h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ n.Add(1)
+ bucket, rawKey, _ := strings.Cut(strings.TrimPrefix(r.URL.EscapedPath(), "/"), "/")
+ key, err := url.PathUnescape(rawKey)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "InvalidURI")
+ return
+ }
+ q := r.URL.Query()
+ if len(allowed) > 0 && !allowed[bucket] {
+ s3err(w, http.StatusMovedPermanently, "PermanentRedirect")
+ return
+ }
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if err := f.known(bucket); err != nil {
+ s3err(w, http.StatusNotFound, "NoSuchBucket")
+ return
+ }
+ switch {
+ case r.Method == http.MethodGet && key == "" && q.Has("versions"):
+ f.serveList(w, bucket, q)
+ case r.Method == http.MethodGet && key != "":
+ body, err := f.read(bucket, key, q.Get("versionId"))
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.Write(body)
+ case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "":
+ f.serveCopy(w, r, bucket, key)
+ case r.Method == http.MethodPut:
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "IncompleteBody") // a failed upload stores nothing
+ return
+ }
+ if r.Header.Get("If-None-Match") != "*" {
+ s3err(w, http.StatusNotImplemented, "NotImplemented") // VersionedStore only puts create-only
+ return
+ }
+ id, err := f.putIfAbsentLocked(bucket, key, body)
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.Header().Set("x-amz-version-id", id)
+ w.Header().Set("ETag", `"`+md5hex(body)+`"`)
+ case r.Method == http.MethodDelete && key != "" && q.Get("versionId") != "":
+ if err := f.deleteLocked(bucket, key, q.Get("versionId")); err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed")
+ }
+ })
+ return h, func() int { return int(n.Load()) }
+}
+
+func (f *VersionedBucket) serveList(w http.ResponseWriter, bucket string, q url.Values) {
+ if err := f.fail("list", q.Get("prefix")); err != nil {
+ writeErr(w, err)
+ return
+ }
+ all := f.listAll(bucket, q.Get("prefix"))
+ km, vm := q.Get("key-marker"), q.Get("version-id-marker")
+ start := 0
+ if km != "" {
+ start = len(all) // a marker not found resumes at the end
+ for i, v := range all {
+ if v.Key == km && v.VersionID == vm {
+ start = i + 1
+ break
+ }
+ }
+ }
+ size := f.PageSize
+ if size <= 0 {
+ size = 1000
+ }
+ page := (start + size - 1) / size // pages already served before this one
+ vs, more, err := f.pageOf(all[start:], 0)
+ if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
+ err = errors.New("leasetest: injected listing failure")
+ }
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ res := listVersionsResult{Name: bucket, Prefix: q.Get("prefix"), KeyMarker: km, VersionIdMarker: vm, MaxKeys: size, IsTruncated: more}
+ for _, v := range vs {
+ lm := v.LastModified.UTC().Format(time.RFC3339)
+ if v.DeleteMarker {
+ res.Entries = append(res.Entries, xmlMarker{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm})
+ } else {
+ res.Entries = append(res.Entries, xmlVersion{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm, ETag: `"` + v.ETag + `"`, Size: v.Size})
+ }
+ }
+ if more && len(vs) > 0 {
+ res.NextKeyMarker, res.NextVersionIdMarker = vs[len(vs)-1].Key, vs[len(vs)-1].VersionID
+ }
+ w.Header().Set("Content-Type", "application/xml")
+ xml.NewEncoder(w).Encode(res)
+}
+
+func (f *VersionedBucket) serveCopy(w http.ResponseWriter, r *http.Request, bucket, key string) {
+ src := strings.TrimPrefix(r.Header.Get("x-amz-copy-source"), "/")
+ srcPath, rawQ, _ := strings.Cut(src, "?")
+ if strings.ContainsAny(srcPath, "+? ") {
+ s3err(w, http.StatusBadRequest, "InvalidArgument") // AWS's encoder escapes these; a raw one is a client bug
+ return
+ }
+ sq, err := url.ParseQuery(rawQ)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "InvalidArgument")
+ return
+ }
+ unesc, err := url.PathUnescape(srcPath)
+ if err != nil {
+ s3err(w, http.StatusBadRequest, "InvalidArgument")
+ return
+ }
+ sb, sk, _ := strings.Cut(unesc, "/")
+ id, etag, err := f.copyLocked(sb, sk, sq.Get("versionId"), bucket, key, r.Header.Get("If-None-Match") == "*")
+ if errors.Is(err, lease.ErrIsDeleteMarker) {
+ s3errMsg(w, http.StatusBadRequest, "InvalidRequest", copyMarkerMsg) // a COPY is 400, not GET's 405
+ return
+ }
+ if err != nil {
+ writeErr(w, err)
+ return
+ }
+ w.Header().Set("x-amz-version-id", id)
+ w.Header().Set("Content-Type", "application/xml")
+ io.WriteString(w, `<CopyObjectResult><ETag>"`+etag+`"</ETag><LastModified>2026-10-05T00:00:00Z</LastModified></CopyObjectResult>`)
+}
diff --git a/internal/lease/leasetest/s3server_test.go b/internal/lease/leasetest/s3server_test.go
new file mode 100644
index 000000000..7ea3b6785
--- /dev/null
+++ b/internal/lease/leasetest/s3server_test.go
@@ -0,0 +1,38 @@
+package leasetest
+
+import (
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+)
+
+// failingReader yields some bytes and then an error, like a dropped upload.
+type failingReader struct{ r io.Reader }
+
+func (f *failingReader) Read(p []byte) (int, error) {
+ n, err := f.r.Read(p)
+ if err == io.EOF {
+ return n, errors.New("connection reset")
+ }
+ return n, err
+}
+
+func TestS3ServerFailedUploadStoresNothing(t *testing.T) {
+ m := NewVersionedBucket()
+ m.AddBucket("data")
+ h, _ := NewS3Server(m)
+ before := m.Snapshot()
+ req := httptest.NewRequest(http.MethodPut, "/data/k", &failingReader{strings.NewReader("partial")})
+ req.Header.Set("If-None-Match", "*")
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ if rec.Code < 400 {
+ t.Fatalf("failed upload answered %d", rec.Code)
+ }
+ if m.Snapshot() != before {
+ t.Fatalf("failed upload changed the bucket:\n%s", m.Snapshot())
+ }
+}
diff --git a/internal/lease/versioned.go b/internal/lease/versioned.go
new file mode 100644
index 000000000..54377d71b
--- /dev/null
+++ b/internal/lease/versioned.go
@@ -0,0 +1,268 @@
+package lease
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/aws/smithy-go"
+ "github.com/aws/smithy-go/encoding/httpbinding"
+ smithyhttp "github.com/aws/smithy-go/transport/http"
+)
+
+// ObjectVersion is one entry of a versioned listing: an object version, or
+// a delete marker (DeleteMarker true, no ETag or Size).
+type ObjectVersion struct {
+ Key string
+ VersionID string
+ ETag string // unquoted
+ Size int64
+ IsLatest bool
+ DeleteMarker bool
+ LastModified time.Time
+}
+
+var (
+ ErrExists = errors.New("lease: object already exists")
+ ErrDestinationExists = errors.New("lease: copy destination already exists")
+ ErrIsDeleteMarker = errors.New("lease: version is a delete marker")
+ ErrNoVersionID = errors.New("lease: a version id is required")
+)
+
+// VersionedStore is the data bucket addressed by exact object version (host
+// lifecycle claim spec C3, §6.2, §6.5). No method deletes or overwrites by
+// bare key.
+type VersionedStore interface {
+ // ListVersions returns every version and delete marker under prefix:
+ // key order; within a key the IsLatest entry first, the rest newest
+ // first with ties unordered. Any page error fails the whole call.
+ ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error)
+ // OpenVersion streams one exact version. Missing: ErrMirrorNotFound.
+ // A delete marker: ErrIsDeleteMarker.
+ OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error)
+ // GetVersion reads one exact version whole (OpenVersion's errors).
+ GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error)
+ // CopyVersion copies one exact source version to dstKey. createOnly
+ // adds If-None-Match: *; an existing current destination object gives
+ // ErrDestinationExists. It returns the destination's new version id
+ // and ETag.
+ CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (dstVersionID, dstETag string, err error)
+ // DeleteVersion removes one exact version; gone already is success.
+ // An empty versionID is ErrNoVersionID, without any request.
+ DeleteVersion(ctx context.Context, bucket, key, versionID string) error
+ // PutIfAbsent writes body only if key holds no current object version.
+ PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
+}
+
+// versionedS3 binds every bucket to its own regional client explicitly.
+// An unbound bucket is an error, never a guess.
+type versionedS3 struct {
+ clients map[string]*awss3.Client // bucket -> client for its region
+}
+
+// NewVersionedS3 builds one client per region from the default credential
+// chain (with s3compat options, as newLeaserClient does) and binds each
+// bucket to its region's client.
+func NewVersionedS3(ctx context.Context, regionByBucket map[string]string) (VersionedStore, error) {
+ byRegion := map[string]*awss3.Client{}
+ out := &versionedS3{clients: map[string]*awss3.Client{}}
+ for bucket, region := range regionByBucket {
+ c, ok := byRegion[region]
+ if !ok {
+ var err error
+ if c, _, err = newLeaserClient(ctx, region); err != nil {
+ return nil, fmt.Errorf("lease: versioned client for %s: %w", region, err)
+ }
+ byRegion[region] = c
+ }
+ out.clients[bucket] = c
+ }
+ return out, nil
+}
+
+func (s *versionedS3) client(bucket string) (*awss3.Client, error) {
+ c, ok := s.clients[bucket]
+ if !ok {
+ return nil, fmt.Errorf("lease: bucket %q is not bound to a region", bucket)
+ }
+ return c, nil
+}
+
+func (s *versionedS3) ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error) {
+ c, err := s.client(bucket)
+ if err != nil {
+ return nil, err
+ }
+ p := awss3.NewListObjectVersionsPaginator(c, &awss3.ListObjectVersionsInput{
+ Bucket: aws.String(bucket), Prefix: aws.String(prefix),
+ }) // StopOnDuplicateToken stays off: a key legitimately repeats across version pages
+ var out []ObjectVersion
+ for p.HasMorePages() {
+ page, err := p.NextPage(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("lease: list versions %s/%s: %w", bucket, prefix, err)
+ }
+ for _, v := range page.Versions {
+ out = append(out, ObjectVersion{Key: aws.ToString(v.Key), VersionID: aws.ToString(v.VersionId),
+ ETag: strings.Trim(aws.ToString(v.ETag), `"`), Size: aws.ToInt64(v.Size),
+ IsLatest: aws.ToBool(v.IsLatest), LastModified: aws.ToTime(v.LastModified)})
+ }
+ for _, d := range page.DeleteMarkers {
+ out = append(out, ObjectVersion{Key: aws.ToString(d.Key), VersionID: aws.ToString(d.VersionId),
+ IsLatest: aws.ToBool(d.IsLatest), DeleteMarker: true, LastModified: aws.ToTime(d.LastModified)})
+ }
+ }
+ sort.SliceStable(out, func(i, j int) bool {
+ if out[i].Key != out[j].Key {
+ return out[i].Key < out[j].Key
+ }
+ if out[i].IsLatest != out[j].IsLatest {
+ return out[i].IsLatest
+ }
+ return out[i].LastModified.After(out[j].LastModified) // ties: unspecified (Global Constraints)
+ })
+ return out, nil
+}
+
+// apiCode returns the smithy error code and HTTP status of err.
+func apiCode(err error) (string, int) {
+ var ae smithy.APIError
+ code := ""
+ if errors.As(err, &ae) {
+ code = ae.ErrorCode()
+ }
+ var re *smithyhttp.ResponseError
+ status := 0
+ if errors.As(err, &re) {
+ status = re.HTTPStatusCode()
+ }
+ return code, status
+}
+
+// mapReadErr normalizes only object or version absence. A missing bucket
+// (NoSuchBucket, also a 404) stays an error: it is never "already gone".
+func mapReadErr(err error) error {
+ code, status := apiCode(err)
+ switch {
+ case code == "NoSuchBucket":
+ return err
+ case code == "NoSuchKey" || code == "NoSuchVersion" || code == "NotFound" || status == http.StatusNotFound:
+ return ErrMirrorNotFound
+ case status == http.StatusMethodNotAllowed:
+ return ErrIsDeleteMarker // S3 answers 405 for a delete marker's version id
+ }
+ return err
+}
+
+func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ if versionID == "" {
+ return nil, ErrNoVersionID
+ }
+ c, err := s.client(bucket)
+ if err != nil {
+ return nil, err
+ }
+ out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)})
+ if err != nil {
+ if m := mapReadErr(err); m != err {
+ return nil, m
+ }
+ return nil, fmt.Errorf("lease: get %s/%s@%s: %w", bucket, key, versionID, err)
+ }
+ return out.Body, nil
+}
+
+func (s *versionedS3) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
+ rc, err := s.OpenVersion(ctx, bucket, key, versionID)
+ if err != nil {
+ return nil, err
+ }
+ defer rc.Close()
+ return io.ReadAll(rc)
+}
+
+func (s *versionedS3) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ if srcVersionID == "" {
+ return "", "", ErrNoVersionID
+ }
+ if _, err := s.client(srcBucket); err != nil {
+ return "", "", err
+ }
+ c, err := s.client(dstBucket) // a copy executes in the destination's region
+ if err != nil {
+ return "", "", err
+ }
+ in := &awss3.CopyObjectInput{
+ Bucket: aws.String(dstBucket),
+ Key: aws.String(dstKey),
+ CopySource: aws.String(httpbinding.EscapePath(srcBucket+"/"+srcKey, false) + "?versionId=" + url.QueryEscape(srcVersionID)),
+ }
+ if createOnly {
+ in.IfNoneMatch = aws.String("*")
+ }
+ out, err := c.CopyObject(ctx, in)
+ if err != nil {
+ code, status := apiCode(err)
+ var ae smithy.APIError
+ switch {
+ case createOnly && (code == "PreconditionFailed" || status == http.StatusPreconditionFailed):
+ return "", "", ErrDestinationExists
+ case code == "InvalidRequest" && errors.As(err, &ae) && strings.Contains(strings.ToLower(ae.ErrorMessage()), "delete marker"):
+ return "", "", ErrIsDeleteMarker
+ default:
+ if m := mapReadErr(err); m != err {
+ return "", "", m
+ }
+ }
+ return "", "", fmt.Errorf("lease: copy %s/%s@%s -> %s/%s: %w", srcBucket, srcKey, srcVersionID, dstBucket, dstKey, err)
+ }
+ etag := ""
+ if out.CopyObjectResult != nil {
+ etag = strings.Trim(aws.ToString(out.CopyObjectResult.ETag), `"`)
+ }
+ return aws.ToString(out.VersionId), etag, nil
+}
+
+func (s *versionedS3) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
+ if versionID == "" {
+ return ErrNoVersionID // never a bare-key delete (spec C3), and no request is sent
+ }
+ c, err := s.client(bucket)
+ if err != nil {
+ return err
+ }
+ if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)}); err != nil {
+ if mapReadErr(err) == ErrMirrorNotFound {
+ return nil // already gone
+ }
+ return fmt.Errorf("lease: delete %s/%s@%s: %w", bucket, key, versionID, err)
+ }
+ return nil
+}
+
+func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ c, err := s.client(bucket)
+ if err != nil {
+ return "", err
+ }
+ out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key),
+ Body: bytes.NewReader(body), IfNoneMatch: aws.String("*")})
+ if err != nil {
+ if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed {
+ return "", ErrExists
+ }
+ return "", fmt.Errorf("lease: put-if-absent %s/%s: %w", bucket, key, err)
+ }
+ return aws.ToString(out.VersionId), nil
+}
+
+var _ VersionedStore = (*versionedS3)(nil)
diff --git a/internal/lease/versioned_test.go b/internal/lease/versioned_test.go
new file mode 100644
index 000000000..a7d415af3
--- /dev/null
+++ b/internal/lease/versioned_test.go
@@ -0,0 +1,97 @@
+package lease_test
+
+import (
+ "context"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+)
+
+func wireClient(t *testing.T, url string) *awss3.Client {
+ t.Helper()
+ return awss3.New(awss3.Options{
+ Region: "eu-west-1", BaseEndpoint: aws.String(url), UsePathStyle: true,
+ Credentials: credentials.NewStaticCredentialsProvider("k", "s", ""),
+ RetryMaxAttempts: 1,
+ })
+}
+
+// wireStore is a versionedS3 whose buckets "data" and "other" are served
+// by two separate servers over one model, so a wrong-region routing is a
+// visible failure.
+func wireStore(t *testing.T) (lease.VersionedStore, *leasetest.VersionedBucket, func() int) {
+ t.Helper()
+ m := leasetest.NewVersionedBucket()
+ m.AddBucket("data")
+ m.AddBucket("other")
+ h1, n1 := leasetest.NewS3Server(m, "data", "nobucket")
+ h2, n2 := leasetest.NewS3Server(m, "other")
+ s1, s2 := httptest.NewServer(h1), httptest.NewServer(h2)
+ t.Cleanup(s1.Close)
+ t.Cleanup(s2.Close)
+ // "nobucket" is bound but absent from the model: the server answers
+ // NoSuchBucket, for the missing-bucket contract case.
+ st := lease.NewVersionedS3WithClients(map[string]*awss3.Client{
+ "data": wireClient(t, s1.URL), "other": wireClient(t, s2.URL),
+ "nobucket": wireClient(t, s1.URL),
+ })
+ return st, m, func() int { return n1() + n2() }
+}
+
+func TestVersionedS3Contract(t *testing.T) {
+ leasetest.RunVersionedStoreContract(t, func(t *testing.T) (lease.VersionedStore, *leasetest.VersionedBucket, string, func() int) {
+ st, m, served := wireStore(t)
+ return st, m, "data", served
+ })
+}
+
+// Each bucket reaches only its own endpoint: a client bound to the wrong
+// endpoint gets 301 and the operation fails.
+func TestVersionedS3RoutesEachBucketToItsEndpoint(t *testing.T) {
+ ctx := context.Background()
+ st, m, _ := wireStore(t)
+ for _, b := range []string{"data", "other"} {
+ v := m.Put(b, "k", []byte(b))
+ if vs, err := st.ListVersions(ctx, b, ""); err != nil || len(vs) != 1 {
+ t.Fatalf("list %s = %v, %v", b, vs, err)
+ }
+ if err := st.DeleteVersion(ctx, b, "k", v); err != nil {
+ t.Fatalf("delete %s = %v", b, err)
+ }
+ }
+ m2 := leasetest.NewVersionedBucket()
+ m2.AddBucket("data")
+ h, _ := leasetest.NewS3Server(m2, "other")
+ srv := httptest.NewServer(h)
+ t.Cleanup(srv.Close)
+ wrong := lease.NewVersionedS3WithClients(map[string]*awss3.Client{"data": wireClient(t, srv.URL)})
+ if _, err := wrong.ListVersions(ctx, "data", ""); err == nil {
+ t.Fatal("a wrong-region endpoint answered")
+ }
+}
+
+func TestVersionedS3RefusesUnboundBucket(t *testing.T) {
+ st, _, _ := wireStore(t)
+ if _, err := st.ListVersions(context.Background(), "unbound", ""); err == nil {
+ t.Fatal("listing an unbound bucket did not fail")
+ }
+}
+
+func TestVersionedS3CopyEscapesLikeAWS(t *testing.T) {
+ st, m, _ := wireStore(t)
+ key := "src/a b+c%?d"
+ v1 := m.Put("data", key, []byte("old"))
+ m.Put("data", key, []byte("new"))
+ if _, _, err := st.CopyVersion(context.Background(), "data", key, v1, "other", "dst/k", true); err != nil {
+ t.Fatal(err)
+ }
+ if cur, ok := m.Current("other", "dst/k"); !ok || string(cur) != "old" {
+ t.Fatalf("copied %q, want the pinned old version", cur)
+ }
+}