amadan

carlos / platform Public

moves: readiness takes only evidence that holds: the logical DB size, no report on an archive retry, the checked release, S1 from the instance itself (Task 14 fix round 1)

- T6 and T11 price Commit x PageSize from the newest staged LTX header,
  or the source's larger local file, never compressed staged bytes.
- archive.ErrChanged ends the round before any report or readiness.
- Ready requires the target's checked release to equal the source's;
  promotion pins it unchanged.
- S1 wakes through the activator and asks the instance over its own
  socket: a 2xx, body read in full, within the deadline; edge
  ServeUncached removed.
- An unreadable instances manifest fails S1 instead of probing "/".
- The archive is sealed to a file under DataDir and staged from it
  (StagingWriter.PutReaderVersionedAt); T5 verdicts are evicted when
  their move ends; MinIO cleanup asserts not-found; the target's helpers
  are split into focused files.
- Copy MV-N148..150, MV-N116 reworded; copy gate literals extended.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude a3a70d455da65bd9fefc33a122874210819d42d3 parent c2312e7
24 files changed, +1228 −564
  • approved_moves_copy_test.go +4 −1
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +4 −1
  • internal/edge/door.go +0 −10
  • internal/edge/edge.go +1 −15
  • internal/edge/edge_test.go +0 −72
  • internal/host/agent.go +0 −3
  • internal/host/movesguard.go +18 −0
  • internal/host/movesheldenv.go +67 −0
  • internal/host/movesmeasure.go +111 −0
  • internal/host/movespass.go +22 −7
  • internal/host/movesphase1.go +75 −34
  • internal/host/movesphase1_harness_test.go +77 −12
  • internal/host/movesphase1_test.go +106 −0
  • internal/host/movesprobe.go +133 −0
  • internal/host/movesprobe_test.go +126 −0
  • internal/host/movesreport.go +121 −295
  • internal/host/movesreport_minio_test.go +6 −2
  • internal/host/movesreport_test.go +86 −23
  • internal/host/movesstage.go +0 −69
  • internal/host/movesstore.go +135 −0
  • internal/lease/staging.go +40 −7
  • internal/lease/staging_test.go +45 −0
  • internal/moves/readiness.go +22 −10
  • internal/moves/readiness_test.go +29 −3
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index fbb3b6661..ebcf37d53 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -64,7 +64,7 @@ var movesCopyIDs = []string{
"MV-N127", "MV-N128", "MV-N129", "MV-N130", "MV-N131", "MV-N132",
"MV-N133", "MV-N134", "MV-N135", "MV-N136", "MV-N137", "MV-N138",
"MV-N139", "MV-N140", "MV-N141", "MV-N142", "MV-N143", "MV-N144",
- "MV-N145", "MV-N146", "MV-N147",
+ "MV-N145", "MV-N146", "MV-N147", "MV-N148", "MV-N149", "MV-N150",
}
// copyRow is one row of the record.
@@ -504,13 +504,16 @@ var movesScanFiles = []string{
"internal/host/movesfloor_linux.go",
"internal/host/movesfloor_other.go",
"internal/host/movesguard.go",
+ "internal/host/movesheldenv.go",
"internal/host/movesmeasure.go",
"internal/host/movesmeasure_linux.go",
"internal/host/movesmeasure_other.go",
"internal/host/movespass.go",
"internal/host/movesphase1.go",
+ "internal/host/movesprobe.go",
"internal/host/movesreport.go",
"internal/host/movesstage.go",
+ "internal/host/movesstore.go",
"internal/host/systemapply.go",
"internal/hostbox/writer.go",
"internal/hostclaim/move.go",
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index 7fe55e22d..c94545d59 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -155,7 +155,7 @@ Columns: id, file, text, status.
| MV-N113 | internal/moves/readiness.go | the archive's file count could not be measured | pending |
| MV-N114 | internal/host/movesphase1.go | this deployment has no deployment bucket for the staging lease | pending |
| MV-N115 | internal/host/movesphase1.go | this box has no secrets key, so it cannot report the launch environment or seal the archive | pending |
-| MV-N116 | internal/host/movesphase1.go | this box's edge is not ready to probe the instance yet | pending |
+| MV-N116 | internal/host/movesphase1.go | this box cannot probe the instance yet | pending |
| MV-N117 | internal/host/movesphase1.go | <host> is not an exec-backed instance, so it cannot move | pending |
| MV-N118 | internal/host/movesphase1.go | <host> does not hibernate, so it cannot move | pending |
| MV-N119 | internal/host/movesphase1.go | <host> has a sidecar environment file, so it cannot move | pending |
@@ -187,6 +187,9 @@ Columns: id, file, text, status.
| MV-N145 | internal/host/movesreport.go | the instance left <region> recently and can move back after <RFC 3339 time> | pending |
| MV-N146 | internal/moves/readiness.go | no fresh rate measurement for the disk's per-file sync (MV-N68 with a new <which>) | pending |
| MV-N147 | internal/console/apiinstancemove.go | invalid region | pending |
+| MV-N148 | internal/host/movesphase1.go | the app's readiness path could not be read | pending |
+| MV-N149 | internal/moves/readiness.go | the target has not checked a release yet | pending |
+| MV-N150 | internal/moves/readiness.go | the target checked release <release> and the source now runs <release> | pending |
## Plan 2
diff --git a/internal/edge/door.go b/internal/edge/door.go
index 9645df948..c9438f5bb 100644
--- a/internal/edge/door.go
+++ b/internal/edge/door.go
@@ -59,16 +59,6 @@ func doorCallerFrom(ctx context.Context) (doorCaller, bool) {
return c, ok
}
-// uncachedProbeKey marks a request ServeUncached is answering: the response
-// cache is neither read nor written for it. A context value, like
-// doorCallerKey, so nothing outside this process can set it.
-type uncachedProbeKey struct{}
-
-func uncachedProbe(ctx context.Context) bool {
- v, _ := ctx.Value(uncachedProbeKey{}).(bool)
- return v
-}
-
// doorOwnsRoute is the whole ownership rule: the route's account and app are
// the caller's.
func doorOwnsRoute(c doorCaller, rt registry.Route) bool {
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 12921e1ad..a4222f36b 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -1341,15 +1341,6 @@ func firstSegmentClean(p string) string {
return firstSegment(path.Clean(p))
}
-// ServeUncached is ServeHTTP with the response cache bypassed in both
-// directions: no cached answer is served and nothing answered is stored,
-// while a hibernating route is still woken through the activator. It is
-// the source's S1 probe (moves spec §7.3, round-4 finding 11): a cached 200
-// must never certify an instance that cannot wake.
-func (e *Edge) ServeUncached(w http.ResponseWriter, r *http.Request) {
- e.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), uncachedProbeKey{}, true)))
-}
-
func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
host := r.Host
if h, _, err := net.SplitHostPort(host); err == nil {
@@ -1657,13 +1648,8 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// instead of being answered from cache, which is the invariant this cache
// exists to hold: the status host caches for itself; its responses never
// enter the shared response cache.
- //
- // A request ServeUncached marks is never cacheable either: a move's
- // readiness probe must hear the instance's own process (a hibernating
- // one is woken below), and what it hears must not be served to anyone
- // else from the cache.
cacheable := e.cache != nil && (r.Method == http.MethodGet || r.Method == http.MethodHead) &&
- up.route.Kind != "status" && !uncachedProbe(r.Context())
+ up.route.Kind != "status"
var key CacheKey
if cacheable {
key = CacheKey{Host: host, Method: r.Method, Path: r.URL.Path, RawQuery: r.URL.RawQuery}
diff --git a/internal/edge/edge_test.go b/internal/edge/edge_test.go
index 0ef9fc46f..2f92f98dc 100644
--- a/internal/edge/edge_test.go
+++ b/internal/edge/edge_test.go
@@ -2324,75 +2324,3 @@ func TestSidecarRouteIsNeverServable(t *testing.T) {
t.Fatalf("GET %s = %d, want 404 — a sidecar must be unreachable", host, code)
}
}
-
-// wakeStub is a hibernation activator that answers every request itself:
-// with the wake failure when fail is set, else with a cacheable 200.
-type wakeStub struct {
- fail bool
- wakes int
-}
-
-func (a *wakeStub) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
- a.wakes++
- if a.fail {
- MarkAnswered(r)
- http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
- return true
- }
- w.Header().Set("Cache-Control", "public, max-age=300")
- w.WriteHeader(http.StatusOK)
- _, _ = w.Write([]byte("woken"))
- return true
-}
-func (a *wakeStub) RetireExcept(map[string]bool) {}
-
-// TestServeUncachedBypassesTheCacheAndWakes: a move's S1 probe (moves spec
-// §7.3, round-4 finding 11) must hear from the instance's own process, never
-// from the response cache. With the cache primed with a 200 for the path and
-// a wake that fails, ServeHTTP answers the cached 200 while ServeUncached
-// wakes and answers the failure; and what ServeUncached answers is never
-// stored, where ServeHTTP would have stored it.
-func TestServeUncachedBypassesTheCacheAndWakes(t *testing.T) {
- reg := openReg(t)
- rt := registry.Route{Host: "sleepy.example.com", Kind: "instance", Socket: t.TempDir() + "/x.sock", Status: "ready", Hibernate: true, App: "web"}
- if err := reg.Upsert(rt); err != nil {
- t.Fatal(err)
- }
- act := &wakeStub{fail: true}
- e, err := New(reg, act)
- if err != nil {
- t.Fatal(err)
- }
- cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
- e.EnableCache(cache)
- cache.Put(CacheKey{Host: rt.Host, Method: http.MethodGet, Path: "/ready"}, "",
- &CachedResponse{StatusCode: http.StatusOK, Header: http.Header{}, Body: []byte("cached"), StoredAt: time.Now(), TTL: time.Hour})
-
- rec := httptest.NewRecorder()
- e.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/ready", nil))
- if rec.Code != http.StatusOK || rec.Body.String() != "cached" || act.wakes != 0 {
- t.Fatalf("ServeHTTP: %d %q after %d wakes, want the cached 200 and no wake", rec.Code, rec.Body.String(), act.wakes)
- }
- rec = httptest.NewRecorder()
- e.ServeUncached(rec, httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/ready", nil))
- if rec.Code != http.StatusServiceUnavailable || act.wakes != 1 {
- t.Fatalf("ServeUncached: %d after %d wakes, want the wake's 503", rec.Code, act.wakes)
- }
-
- // A wake that answers a cacheable 200: ServeUncached stores nothing.
- act.fail = false
- rec = httptest.NewRecorder()
- e.ServeUncached(rec, httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/fresh", nil))
- if rec.Code != http.StatusOK || rec.Body.String() != "woken" {
- t.Fatalf("ServeUncached: %d %q", rec.Code, rec.Body.String())
- }
- probe := httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/fresh", nil)
- if _, hit := cache.GetFor(CacheKey{Host: rt.Host, Method: http.MethodGet, Path: "/fresh"}, probe); hit {
- t.Fatal("ServeUncached stored its answer in the response cache")
- }
- // The control: ServeHTTP stores the same answer.
- e.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "http://sleepy.example.com/fresh", nil))
- if _, hit := cache.GetFor(CacheKey{Host: rt.Host, Method: http.MethodGet, Path: "/fresh"}, probe); !hit {
- t.Fatal("the control did not store: the cacheable answer is not cacheable, so the test proves nothing")
- }
-}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 49aef548c..59c71f5bf 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -612,9 +612,6 @@ func runAgent(args []string, buildClaims ClaimFactory) {
edgeDoorOptions(os.Getenv, readPublicIP)...)
edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
go tunnelLoop(e)
- // A Phase 1 move's S1 probe asks through this edge, uncached
- // (moves design §7.3); before this it answers "edge not ready".
- setMovesReadyProbe(e)
// The edge owns the health probers, so its state table is only
// reachable once it exists — statusLoop (already running above) reads
// it through this published source on its next tick. Method value, so
diff --git a/internal/host/movesguard.go b/internal/host/movesguard.go
index 60bedba55..8197949b3 100644
--- a/internal/host/movesguard.go
+++ b/internal/host/movesguard.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "os"
"time"
"github.com/carlosframework/platform/internal/lease"
@@ -70,6 +71,23 @@ func (m *guardedMirror) PutVersioned(ctx context.Context, bucket, key string, bo
return versionID, err
}
+// PutFileVersioned stages the file at path (size bytes, sha256Hex) as key,
+// read from disk inside the guarded call: a retry opens it again under a
+// new check.
+func (m *guardedMirror) PutFileVersioned(ctx context.Context, bucket, key, path string, size int64, sha256Hex string) (versionID string, err error) {
+ err = m.g.Before(func(signAt time.Time) error {
+ f, oerr := os.Open(path)
+ if oerr != nil {
+ return oerr
+ }
+ defer f.Close()
+ var perr error
+ versionID, perr = m.w.PutReaderVersionedAt(ctx, signAt, bucket, key, f, size, sha256Hex)
+ return perr
+ })
+ return versionID, err
+}
+
// PutIfAbsentVersioned is move.json's create-only write; an existing object
// is lease.ErrExists.
func (m *guardedMirror) PutIfAbsentVersioned(ctx context.Context, bucket, key string, body []byte) (versionID string, err error) {
diff --git a/internal/host/movesheldenv.go b/internal/host/movesheldenv.go
new file mode 100644
index 000000000..d4bf651e7
--- /dev/null
+++ b/internal/host/movesheldenv.go
@@ -0,0 +1,67 @@
+package host
+
+import (
+ "context"
+ "crypto/ecdh"
+ "os"
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// T3's production composition of a flat held row (moves design §7.3,
+// ruling F19). armPhase1 wires it as moveCfg.ComposeHeldEnv.
+
+// composeHeldEnv is T3's composition for a flat held row (ruling F19): a
+// scratch directory holding the configuration rendered from the merged
+// bundle (appconfig.LoadMergedHostVersions, rendered as RenderHeld renders
+// it) and the carried <host>.env, an ExecBackend over it with this box's
+// region and cache, and a synthetic flat route; LaunchEnvironment composes
+// what the next spawn would get.
+func composeHeldEnv(priv *ecdh.PrivateKey, region string, cache bool) func(context.Context, release.Store, provision.Record, []byte) (map[string]string, []string, provision.MoveInputs, error) {
+ return func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, []string, provision.MoveInputs, error) {
+ m, v, err := appconfig.LoadMergedHostVersions(ctx, s, rec.Account, rec.App, rec.Environment, rec.Host)
+ if err != nil {
+ return nil, nil, provision.MoveInputs{}, err
+ }
+ var rendered []byte
+ if m != nil {
+ if rendered, err = appconfig.RenderEnv(m, priv, appconfig.Binding{Account: rec.Account, App: rec.App}); err != nil {
+ return nil, nil, provision.MoveInputs{}, err
+ }
+ }
+ tmp, err := os.MkdirTemp("", "carlos-move-t3-")
+ if err != nil {
+ return nil, nil, provision.MoveInputs{}, err
+ }
+ defer os.RemoveAll(tmp)
+ if err := os.WriteFile(filepath.Join(tmp, rec.Host+".config.env"), rendered, 0o600); err != nil {
+ return nil, nil, provision.MoveInputs{}, err
+ }
+ if carried != nil {
+ if err := os.WriteFile(filepath.Join(tmp, rec.Host+".env"), carried, 0o600); err != nil {
+ return nil, nil, provision.MoveInputs{}, err
+ }
+ }
+ eb := &activator.ExecBackend{DataDir: tmp, Region: region}
+ if cache {
+ eb.Cache = heldEnvCache
+ }
+ env, shadowed, err := eb.LaunchEnvironment(registry.Route{Host: rec.Host, Socket: provision.DefaultSocket(rec.Host)})
+ if err != nil {
+ return nil, nil, provision.MoveInputs{}, err
+ }
+ in := provision.MoveInputs{BundleETags: map[string]string{"default": v.Default, "environment": v.Environment}, OverlayETag: v.Overlay}
+ return env, shadowed, in, nil
+ }
+}
+
+// heldEnvCache is the non-nil cache T3's synthetic backend carries when the
+// live box has a response cache: LaunchEnvironment reads only whether one
+// is set (CARLOS_PURGE).
+var heldEnvCache = edge.NewCache(1, 1)
diff --git a/internal/host/movesmeasure.go b/internal/host/movesmeasure.go
index 57ad5160b..fd4232c50 100644
--- a/internal/host/movesmeasure.go
+++ b/internal/host/movesmeasure.go
@@ -1,10 +1,15 @@
package host
import (
+ "context"
+ "fmt"
+ "log"
"os"
"path/filepath"
+ "strconv"
"time"
+ "github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
)
@@ -39,3 +44,109 @@ func measureDiskSynced(dir string, now func() time.Time) (moves.MeasuredRate, er
}
return moves.RateOf(2*measureDiskSize, now().Sub(start), now().UTC()), nil
}
+
+// measureS3 times a 16 MiB put and get of the probe object (moves.ProbeKey,
+// outside moves/, F11) in this box's own bucket with its own credentials,
+// and deletes it.
+func measureS3(mirror func(ctx context.Context) (lease.MirrorStore, error), bucket string, now func() time.Time) func(context.Context, string) (moves.MeasuredRate, error) {
+ return func(ctx context.Context, key string) (moves.MeasuredRate, error) {
+ m, err := mirror(ctx)
+ if err != nil {
+ return moves.MeasuredRate{}, err
+ }
+ const size = 16 << 20
+ body := make([]byte, size)
+ start := now()
+ if err := m.Put(ctx, bucket, key, body); err != nil {
+ return moves.MeasuredRate{}, err
+ }
+ got, err := m.Get(ctx, bucket, key)
+ elapsed := now().Sub(start)
+ if derr := m.Delete(ctx, bucket, key); derr != nil {
+ log.Printf("moves: delete the S3 probe object %s: %v", key, derr)
+ }
+ if err != nil {
+ return moves.MeasuredRate{}, err
+ }
+ if len(got) != size {
+ return moves.MeasuredRate{}, fmt.Errorf("the probe object read back %d bytes of %d", len(got), size)
+ }
+ return moves.RateOf(2*size, elapsed, now().UTC()), nil
+ }
+}
+
+// measureDiskIn measures this box's disk under <dataDir>/.moves (the
+// platform-specific write and read are movesmeasure_*.go).
+func measureDiskIn(dataDir string, now func() time.Time) func(context.Context) (moves.MeasuredRate, moves.FilesRate, error) {
+ return func(ctx context.Context) (moves.MeasuredRate, moves.FilesRate, error) {
+ dir, err := os.MkdirTemp(filepath.Join(dataDir, ".moves"), "measure-")
+ if err != nil {
+ if err := os.MkdirAll(filepath.Join(dataDir, ".moves"), 0o755); err != nil {
+ return moves.MeasuredRate{}, moves.FilesRate{}, err
+ }
+ if dir, err = os.MkdirTemp(filepath.Join(dataDir, ".moves"), "measure-"); err != nil {
+ return moves.MeasuredRate{}, moves.FilesRate{}, err
+ }
+ }
+ defer os.RemoveAll(dir)
+ disk, err := measureDiskRate(dir, now)
+ if err != nil {
+ return moves.MeasuredRate{}, moves.FilesRate{}, err
+ }
+ files, err := measureFsyncRate(dir, now)
+ if err != nil {
+ return moves.MeasuredRate{}, moves.FilesRate{}, err
+ }
+ return disk, files, nil
+ }
+}
+
+// measureFsyncFiles and measureFsyncWorkers are the per-file sync probe:
+// a few hundred small files synced by the extract's 16-worker pattern
+// (ruling T7-budget).
+const (
+ measureFsyncFiles = 512
+ measureFsyncWorkers = 16
+)
+
+func measureFsyncRate(dir string, now func() time.Time) (moves.FilesRate, error) {
+ jobs := make(chan int)
+ errs := make(chan error, measureFsyncWorkers)
+ start := now()
+ for i := 0; i < measureFsyncWorkers; i++ {
+ go func() {
+ var first error
+ for n := range jobs {
+ if first != nil {
+ continue
+ }
+ f, err := os.OpenFile(filepath.Join(dir, "f"+strconv.Itoa(n)), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
+ if err == nil {
+ _, err = f.Write(make([]byte, 4096))
+ if err == nil {
+ err = f.Sync()
+ }
+ if cerr := f.Close(); err == nil {
+ err = cerr
+ }
+ }
+ first = err
+ }
+ errs <- first
+ }()
+ }
+ for n := 0; n < measureFsyncFiles; n++ {
+ jobs <- n
+ }
+ close(jobs)
+ for i := 0; i < measureFsyncWorkers; i++ {
+ if err := <-errs; err != nil {
+ return moves.FilesRate{}, err
+ }
+ }
+ d := now().Sub(start)
+ if d <= 0 {
+ d = time.Microsecond
+ }
+ return moves.FilesRate{PerS: measureFsyncFiles / d.Seconds(), At: now().UTC(), Files: measureFsyncFiles}, nil
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 6f51188a4..866dd4d61 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -213,7 +213,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
}
w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
if cfg.Enabled {
- armPhase1(&cfg, w, hib, dataDir, reg, eb, os.Getenv)
+ armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, os.Getenv)
}
w.cfg = cfg
armedMovesWorker.Store(w)
@@ -221,7 +221,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
// armPhase1 wires the Phase 1 hooks (moves design §7.3). Only with the
// switch on: off, none is set and eb.Strict stays nil.
-func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir string, reg *registry.Registry, eb *activator.ExecBackend, getenv func(string) string) {
+func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, getenv func(string) string) {
cfg.Box = metricsBoxLabel(getenv, readMachineID())
cfg.OptDir = release.OptDir()
cfg.DeployBucket = strings.TrimSpace(getenv("CARLOS_DEPLOYMENT_BUCKET"))
@@ -252,7 +252,6 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
cfg.FilesDir = func(host string) (string, string) {
return filepath.Join(dataDir, host), filepath.Join(dataDir, host+".env")
}
- cfg.ReadyProbe = probeThroughEdge
cfg.DiskFree = diskBytes
cfg.ProbeStore = probeStores
cfg.LaneProbe = laneProbe(cfg.Versioned)
@@ -260,6 +259,16 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
cfg.MeasureS3 = measureS3(func(ctx context.Context) (lease.MirrorStore, error) { return w.mirror(ctx, hib.Region) }, hib.Bucket, time.Now)
if reg != nil {
cfg.ReleaseReady = heldReleaseReady(reg, cfg.OptDir, dataDir, cfg.Box, stageRequester(dataDir))
+ if svc != nil {
+ // S1 asks the instance itself, woken through the activator.
+ cfg.ReadyProbe = func(ctx context.Context, host, path string) moves.Check {
+ rt, err := reg.Get(host)
+ if err != nil || rt == nil {
+ return moves.Check{Reason: s1NoAnswer(host)}
+ }
+ return probeInstance(ctx, svc.ServeScheduled, *rt, path)
+ }
+ }
}
if eb != nil {
cfg.StrictNext = eb.NextSpawnStrict
@@ -374,8 +383,8 @@ type moveCfg struct {
// FilesDir is the files directory and <host>.env the archive carries
// (flat: <DataDir>/<host>/ and <DataDir>/<host>.env).
FilesDir func(host string) (dir, envPath string)
- // ReadyProbe is S1: GET path on host through the edge, uncached and
- // waking (setMovesReadyProbe). StrictNext is S2
+ // ReadyProbe is S1: GET path from the instance itself, woken through
+ // the activator (probeInstance). StrictNext is S2
// (ExecBackend.NextSpawnStrict).
ReadyProbe func(ctx context.Context, host, path string) moves.Check
StrictNext func(host string) moves.Check
@@ -508,8 +517,14 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
}
w.trouble.Delete(rec.Host)
}
- if cfg.Enabled && hasPhase1Into(recs, cfg.Region) && loadFleet() {
- w.targetReportPass(ctx, s, recs, *fleet)
+ switch {
+ case !cfg.Enabled:
+ case hasPhase1Into(recs, cfg.Region):
+ if loadFleet() {
+ w.targetReportPass(ctx, s, recs, *fleet)
+ }
+ default:
+ w.evictArchives(nil) // no move into this box: no verdict to keep
}
w.liftFences(recs)
return nil
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go
index 2876874bf..e43a22acc 100644
--- a/internal/host/movesphase1.go
+++ b/internal/host/movesphase1.go
@@ -41,7 +41,15 @@ const (
s1NoDeployBucket = "this deployment has no deployment bucket for the staging lease"
s1Nested = "nested instances cannot move yet"
s1NoSecretsKey = "this box has no secrets key, so it cannot report the launch environment or seal the archive"
- s1EdgeNotReady = "this box's edge is not ready to probe the instance yet"
+ s1CannotProbe = "this box cannot probe the instance yet"
+ s1NoReadyPath = "the app's readiness path could not be read"
+)
+
+// archivePlainDigest and archiveBuild are the archive's digest and build,
+// as variables so a test can make a file change under the walk.
+var (
+ archivePlainDigest = archive.PlainDigest
+ archiveBuild = archive.Build
)
func s1NotExec(host string) string {
@@ -306,15 +314,23 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p
}
sr := &srcRound{deltaBytes: round.DeltaBytes()}
- // The files archive, rebuilt only when its plaintext digest changes.
- if why, err := w.publishArchive(ctx, gm, rec, tgt.DataBucket, &ledger); err != nil {
+ // The files archive, rebuilt only when its plaintext digest changes. A
+ // file that changed under the walk ends the round here, before any
+ // report or readiness: the archive the target checked is no longer the
+ // files, and the next round builds the one that is (review finding 2).
+ why, retry, err := w.publishArchive(ctx, gm, rec, tgt.DataBucket, &ledger)
+ if err != nil {
return err
- } else if why != "" {
- sr.s1 = &moves.Check{Reason: why}
}
if err := saveLedger(); err != nil {
return err
}
+ if retry {
+ return nil
+ }
+ if why != "" {
+ sr.s1 = &moves.Check{Reason: why}
+ }
sr.ledger = ledger
if err := w.writeSourceReport(ctx, s, rec, sr); err != nil {
return err
@@ -439,67 +455,84 @@ func containsString(xs []string, x string) bool {
// publishArchive builds, seals and stages the files archive when its
// plaintext digest differs from the one the ledger published, and records
// the new one. It returns an S1 reason for an archive this move can never
-// carry; a file that changed while it was read is a retry next round, and
-// the archive published before stays.
-func (w *movesWorker) publishArchive(ctx context.Context, gm *guardedMirror, rec provision.Record, dstBucket string, ledger *moveLedger) (string, error) {
+// carry, or retry for a file that changed while it was read: the round
+// then ends, and the next one tries again.
+func (w *movesWorker) publishArchive(ctx context.Context, gm *guardedMirror, rec provision.Record, dstBucket string, ledger *moveLedger) (why string, retry bool, err error) {
cfg := w.cfg
dir, envPath := cfg.FilesDir(rec.Host)
- plain, _, _, err := archive.PlainDigest(dir, envPath)
+ plain, _, _, err := archivePlainDigest(dir, envPath)
if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry {
- return why, ferr
+ return why, retry, ferr
}
if plain == ledger.ArchivePlain && ledger.ArchiveVersions != nil {
- return "", nil
+ return "", false, nil
}
- tmp, err := os.MkdirTemp("", "carlos-move-archive-")
+ // Built and sealed to files on the data disk, and staged from the
+ // sealed file, so a 256 MiB archive is never held in memory.
+ tmp, err := movesScratch(cfg.DataDir, "archive-")
if err != nil {
- return "", err
+ return "", false, err
}
defer os.RemoveAll(tmp)
built, err := os.Create(filepath.Join(tmp, "archive.tar.zst"))
if err != nil {
- return "", err
+ return "", false, err
}
defer built.Close()
- plain, entries, bytes, err := archive.Build(dir, envPath, built)
+ plain, entries, bytes, err := archiveBuild(dir, envPath, built)
if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry {
- return why, ferr
+ return why, retry, ferr
}
if _, err := built.Seek(0, io.SeekStart); err != nil {
- return "", err
+ return "", false, err
}
sealedPath := filepath.Join(tmp, "archive.sealed")
sealed, err := os.Create(sealedPath)
if err != nil {
- return "", err
+ return "", false, err
}
defer sealed.Close()
+ // Seal hashes what it writes: sealedSHA is the sealed file's sha256,
+ // the payload hash its upload is signed with.
dataKey, sealedSHA, err := archive.Seal(built, cfg.SecretsPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealed)
if err != nil {
- return "", fmt.Errorf("seal the archive: %w", err)
+ return "", false, fmt.Errorf("seal the archive: %w", err)
+ }
+ if err := sealed.Sync(); err != nil {
+ return "", false, err
+ }
+ st, err := sealed.Stat()
+ if err != nil {
+ return "", false, err
}
m := archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Incarnation: rec.Move.Incarnation, Plain: plain,
Sealed: sealedSHA, Bytes: bytes, Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize}
manifest, err := json.Marshal(m)
if err != nil {
- return "", err
- }
- sealedBody, err := os.ReadFile(sealedPath)
- if err != nil {
- return "", err
+ return "", false, err
}
name := "files-" + digestHex(sealedSHA)
prefix := moves.FilesPrefix(rec.Host, rec.Move.ID)
ref := &moves.ArchiveRef{ManifestKey: prefix + name + ".json", SealedObjectKey: prefix + name + ".sealed", Plain: plain}
- if ref.SealedVersion, err = gm.PutVersioned(ctx, dstBucket, ref.SealedObjectKey, sealedBody); err != nil {
- return "", fmt.Errorf("stage the sealed archive: %w", err)
+ if ref.SealedVersion, err = gm.PutFileVersioned(ctx, dstBucket, ref.SealedObjectKey, sealedPath, st.Size(), digestHex(sealedSHA)); err != nil {
+ return "", false, fmt.Errorf("stage the sealed archive: %w", err)
}
if ref.ManifestVersion, err = gm.PutVersioned(ctx, dstBucket, ref.ManifestKey, manifest); err != nil {
- return "", fmt.Errorf("stage the archive manifest: %w", err)
+ return "", false, fmt.Errorf("stage the archive manifest: %w", err)
}
ledger.ArchivePlain, ledger.ArchiveSealed, ledger.ArchiveVersions = plain, sealedSHA, ref
log.Printf("moves: %s: staged the files archive %s (%d files, %d bytes)", rec.Host, plain, len(entries), bytes)
- return "", nil
+ return "", false, nil
+}
+
+// movesScratch is a scratch directory under <dataDir>/.moves: on the data
+// disk, never a shared tmpfs.
+func movesScratch(dataDir, pattern string) (string, error) {
+ parent := filepath.Join(dataDir, ".moves")
+ if err := os.MkdirAll(parent, 0o755); err != nil {
+ return "", err
+ }
+ return os.MkdirTemp(parent, pattern)
}
// archiveFailure maps an archive error to an S1 reason (a standing refusal
@@ -556,6 +589,9 @@ func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, re
// NextSpawnStrict prefixes its reasons with "S2: ", which Ready adds.
sr.S2.Reason = strings.TrimPrefix(sr.S2.Reason, "S2: ")
}
+ if st, err := os.Stat(provision.DefaultDBPath(cfg.DataDir, rec.Host)); err == nil && st.Mode().IsRegular() {
+ sr.DBBytes = st.Size()
+ }
if cfg.OptDir != "" {
if v, err := release.LiveVersion(cfg.OptDir, rec.Host); err == nil {
sr.Release = v
@@ -584,13 +620,17 @@ func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, re
// woken if it sleeps, never from the response cache.
func (w *movesWorker) probeSource(ctx context.Context, s release.Store, rec provision.Record) moves.Check {
if w.cfg.ReadyProbe == nil {
- return moves.Check{Reason: s1EdgeNotReady}
+ return moves.Check{Reason: s1CannotProbe}
}
- path := "/"
- if m, err := provision.GetManifest(ctx, s, rec.Account, rec.App); err == nil && m != nil {
- path = m.ReadyPathOrDefault()
+ // "/" only when the manifest was read and declares no path: an app
+ // whose declared path fails must not pass on a healthy homepage while
+ // its manifest cannot be read (review finding 5).
+ m, err := provision.GetManifest(ctx, s, rec.Account, rec.App)
+ if err != nil {
+ log.Printf("moves: %s: S1: instances manifest: %v", rec.Host, err)
+ return moves.Check{Reason: s1NoReadyPath}
}
- return w.cfg.ReadyProbe(ctx, rec.Host, path)
+ return w.cfg.ReadyProbe(ctx, rec.Host, m.ReadyPathOrDefault())
}
// decideReady reads the target's report and applies §7.3's rule; on ready
@@ -614,8 +654,9 @@ func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec prov
w.notice(rec.Host+"#ready", "moves: %s: not ready: %s", rec.Host, why)
return nil
}
+ // moves.Ready has required target.Inputs.Release to equal the
+ // source's, so what is pinned is what T7 checked (review finding 3).
in := target.Inputs
- in.Release = source.Release
if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedPromoteReady(g, cfg.Now(), in)); err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the record no longer holds move %s as preparing (%v); this round ends", rec.Host, g.ID, err)
diff --git a/internal/host/movesphase1_harness_test.go b/internal/host/movesphase1_harness_test.go
index 35486bd7c..67eca74de 100644
--- a/internal/host/movesphase1_harness_test.go
+++ b/internal/host/movesphase1_harness_test.go
@@ -3,6 +3,8 @@ package host
import (
"bytes"
"context"
+ "crypto/sha256"
+ "encoding/hex"
"errors"
"fmt"
"io"
@@ -20,6 +22,7 @@ import (
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
)
// The Phase 1 harness: the legacy moveHarness, its fake mirror extended
@@ -101,6 +104,20 @@ func (f *fakeMoveMirror) PutVersionedAt(_ context.Context, signAt time.Time, buc
return f.putAt("put", signAt, bucket, key, body, false)
}
+func (f *fakeMoveMirror) PutReaderVersionedAt(_ context.Context, signAt time.Time, bucket, key string, r io.Reader, size int64, sha256Hex string) (string, error) {
+ body, err := io.ReadAll(io.LimitReader(r, size+1))
+ if err != nil {
+ return "", err
+ }
+ if int64(len(body)) != size {
+ return "", fmt.Errorf("fake: body is %d bytes, declared %d", len(body), size)
+ }
+ if sum := sha256.Sum256(body); hex.EncodeToString(sum[:]) != sha256Hex {
+ return "", errors.New("fake: XAmzContentSHA256Mismatch")
+ }
+ return f.putAt("put", signAt, bucket, key, body, false)
+}
+
func (f *fakeMoveMirror) PutIfAbsentVersionedAt(_ context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) {
return f.putAt("put-if-absent", signAt, bucket, key, body, true)
}
@@ -263,15 +280,16 @@ var p1Created = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
type phase1Harness struct {
*moveHarness
- vs *fakeVersioned
- lease *fakeStagingLease
- clock *stepClock
- inc string
- route registry.Route
- env map[string]string // the source's launch environment
- optDir string
- probe moves.Check
- strict moves.Check
+ vs *fakeVersioned
+ lease *fakeStagingLease
+ clock *stepClock
+ inc string
+ route registry.Route
+ env map[string]string // the source's launch environment
+ optDir string
+ probe moves.Check
+ probePaths []string
+ strict moves.Check
// worker1 is the source box's worker and worker2 the target's, each
// kept across passes as one agent keeps its own.
worker1, worker2 *movesWorker
@@ -325,7 +343,10 @@ func newPhase1Harness(t *testing.T) *phase1Harness {
h.cfg.FilesDir = func(host string) (string, string) {
return filepath.Join(h.cfg.DataDir, host), filepath.Join(h.cfg.DataDir, host+".env")
}
- h.cfg.ReadyProbe = func(context.Context, string, string) moves.Check { return h.probe }
+ h.cfg.ReadyProbe = func(_ context.Context, _, path string) moves.Check {
+ h.probePaths = append(h.probePaths, path)
+ return h.probe
+ }
h.cfg.StrictNext = func(string) moves.Check { return h.strict }
h.cfg.Route = func(host string) (*registry.Route, error) {
if host != p1Host {
@@ -350,8 +371,10 @@ func newPhase1Harness(t *testing.T) *phase1Harness {
}
pointLiveForTest(t, h.optDir, p1Host, "v7")
loc := "instances/" + p1Host + "/" + inc + "/"
- h.f.putSized("data-ie", loc+"g-cur/0000/0000000000000001-0000000000000002.ltx", "cur-a", 2<<20)
- h.f.putSized("data-ie", loc+"g-cur/0001/0000000000000001-0000000000000002.ltx", "cur-b", 4096)
+ h.f.store("data-ie", loc+"g-cur/0000/0000000000000001-0000000000000002.ltx",
+ fakeMoveObject{lease.MirrorObject{ETag: "cur-a", Size: 2 << 20}, string(ltxForTest(t, 4096, 4, 1, 2))})
+ h.f.store("data-ie", loc+"g-cur/0001/0000000000000001-0000000000000002.ltx",
+ fakeMoveObject{lease.MirrorObject{ETag: "cur-b", Size: 4096}, string(ltxForTest(t, 4096, 4, 1, 2))})
h.f.putSized("data-ie", loc+"g-old/0000/0000000000000001-0000000000000001.ltx", "old-a", 4096)
h.f.put("data-ie", loc+"generation.json", "gen")
h.f.put("data-ie", loc+"g-cur/lock.json", "lock")
@@ -361,6 +384,10 @@ func newPhase1Harness(t *testing.T) *phase1Harness {
[]byte(`{"v":1,"box":"ie-box","eip":"1.2.3.5","placement":"ie","data_bucket":"data-ie","data_bucket_region":"eu-west-1","hosts":{}}`), ""); err != nil {
t.Fatal(err)
}
+ // The app's instances manifest: it declares no readiness path.
+ if err := provision.PutManifest(ctx, h.s, "bab", "app", provision.Manifest{Schema: 1}); err != nil {
+ t.Fatal(err)
+ }
// The console's record (T2).
if _, err := h.s.Put(ctx, moves.ConsoleKey, []byte(`{"v":1,"protocol":1,"secrets_key_fp":"`+appconfig.KeyFingerprint(priv.PublicKey())+`"}`), ""); err != nil {
t.Fatal(err)
@@ -511,3 +538,41 @@ func (h *phase1Harness) ledger(t *testing.T) moveLedger {
}
return l
}
+
+// ltxForTest is a real LTX file (lz4-compressed pages) for a database of
+// commit pages of pageSize bytes, covering minTX to maxTX. A snapshot
+// (minTX 1) carries every page; an increment carries page 1 alone. Pages
+// are zeros, so the file is far smaller than the database it describes.
+func ltxForTest(t *testing.T, pageSize, commit uint32, minTX, maxTX uint64) []byte {
+ t.Helper()
+ var buf bytes.Buffer
+ enc, err := ltx.NewEncoder(&buf)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hdr := ltx.Header{Version: ltx.Version, PageSize: pageSize, Commit: commit, MinTXID: ltx.TXID(minTX), MaxTXID: ltx.TXID(maxTX), Timestamp: 1000}
+ if minTX > 1 {
+ hdr.PreApplyChecksum = ltx.ChecksumFlag | 1
+ }
+ if err := enc.EncodeHeader(hdr); err != nil {
+ t.Fatal(err)
+ }
+ pages := uint32(1)
+ if minTX == 1 {
+ pages = commit
+ }
+ page := make([]byte, pageSize)
+ for pgno := uint32(1); pgno <= pages; pgno++ {
+ if pgno == ltx.LockPgno(pageSize) {
+ continue
+ }
+ if err := enc.EncodePage(ltx.PageHeader{Pgno: pgno}, page); err != nil {
+ t.Fatal(err)
+ }
+ }
+ enc.SetPostApplyChecksum(ltx.ChecksumFlag | 2)
+ if err := enc.Close(); err != nil {
+ t.Fatal(err)
+ }
+ return buf.Bytes()
+}
diff --git a/internal/host/movesphase1_test.go b/internal/host/movesphase1_test.go
index 98b606248..31a873d35 100644
--- a/internal/host/movesphase1_test.go
+++ b/internal/host/movesphase1_test.go
@@ -3,6 +3,8 @@ package host
import (
"context"
"errors"
+ "fmt"
+ "io"
"os"
"path/filepath"
"sort"
@@ -14,6 +16,7 @@ import (
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
"github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
@@ -573,6 +576,50 @@ func TestArchiveIsRebuiltOnlyWhenPlainChanges(t *testing.T) {
if l := h.ledger(t); l.ArchivePlain == first || l.ArchiveVersions == nil || l.ArchiveVersions.SealedVersion == "" || l.ArchiveVersions.ManifestVersion == "" {
t.Fatalf("ledger after the change: %+v", l)
}
+ // A file that changes under the walk (archive.ErrChanged), while the
+ // digest is taken or while the archive is built, after an earlier
+ // archive passed T5: the round ends before readiness, publishes no
+ // source report, and never promotes on the archive the target checked
+ // before (Task 14 review finding 2).
+ for _, stage := range []string{"digest", "build"} {
+ t.Run("changed during the "+stage, func(t *testing.T) {
+ h2 := newPhase1Harness(t)
+ h2.start(t, "manual")
+ h2.pass(t)
+ h2.targetPass(t, h2.targetCfg())
+ if f := failing(h2.targetReport(t)); f[moves.CheckArchive] != "" {
+ t.Fatalf("the first archive did not pass T5: %v", f)
+ }
+ before := h2.sourceReport(t)
+ if err := os.WriteFile(filepath.Join(h2.cfg.DataDir, p1Host, "upload.txt"), []byte("changing"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ changed := fmt.Errorf("%w: upload.txt", archive.ErrChanged)
+ prevDigest, prevBuild := archivePlainDigest, archiveBuild
+ t.Cleanup(func() { archivePlainDigest, archiveBuild = prevDigest, prevBuild })
+ if stage == "digest" {
+ archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, changed }
+ } else {
+ archiveBuild = func(string, string, io.Writer) (string, []archive.Entry, int64, error) { return "", nil, 0, changed }
+ }
+ h2.pass(t)
+ if after := h2.sourceReport(t); !after.At.Equal(before.At) {
+ t.Fatalf("a retried round published a source report (%v, was %v)", after.At, before.At)
+ }
+ if got := h2.read(t, p1Host).Move.State; got != provision.MovePreparing {
+ t.Fatalf("a retried round promoted on the old archive: %s", got)
+ }
+ if l := h2.ledger(t); l.ArchivePlain != before.FilesPlain {
+ t.Fatalf("the ledger moved to %s", l.ArchivePlain)
+ }
+ archivePlainDigest, archiveBuild = prevDigest, prevBuild
+ h2.pass(t)
+ if sr := h2.sourceReport(t); sr.FilesPlain == before.FilesPlain || h2.read(t, p1Host).Move.State != provision.MovePreparing {
+ t.Fatalf("after the retry: files %s, state %s", sr.FilesPlain, h2.read(t, p1Host).Move.State)
+ }
+ })
+ }
+
// An archive this move can never carry is a standing S1 refusal.
if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil {
t.Fatal(err)
@@ -605,3 +652,62 @@ func TestAutoCutMarksCuttingFromReady(t *testing.T) {
t.Fatalf("manual cut: %+v", got)
}
}
+
+// Promotion pins only the release the target checked (Task 14 review
+// finding 3): after the target checks v7, a source that moves to v8 does
+// not promote until the target has checked v8, and then pins v8.
+func TestPromotionNeedsTheReleaseTheTargetChecked(t *testing.T) {
+ h := newPhase1Harness(t)
+ h.start(t, "manual")
+ h.pass(t)
+ h.targetPass(t, h.targetCfg())
+ if r := h.targetReport(t); r.Inputs.Release != "v7" {
+ t.Fatalf("the target checked %q", r.Inputs.Release)
+ }
+ pointLiveForTest(t, h.optDir, p1Host, "v8")
+ h.pass(t)
+ if got := h.read(t, p1Host).Move; got.State != provision.MovePreparing || got.Inputs != nil {
+ t.Fatalf("promoted with v8 pinned on a v7 check: %+v", got)
+ }
+ h.targetPass(t, h.targetCfg())
+ h.pass(t)
+ if got := h.read(t, p1Host).Move; got.State != provision.MoveReady || got.Inputs == nil || got.Inputs.Release != "v8" {
+ t.Fatalf("after the target checked v8: %+v", got)
+ }
+}
+
+// S1 asks the path the instances manifest declares. A manifest that cannot
+// be read fails S1, even though "/" would answer (Task 14 review finding
+// 5); "/" is used only when the manifest was read and declares no path.
+func TestS1ReadsTheDeclaredReadinessPath(t *testing.T) {
+ ctx := context.Background()
+ h := newPhase1Harness(t)
+ h.start(t, "manual")
+ h.pass(t)
+ if sr := h.sourceReport(t); !sr.S1.OK || len(h.probePaths) != 1 || h.probePaths[0] != "/" {
+ t.Fatalf("no declared path: S1 %+v, probed %v", sr.S1, h.probePaths)
+ }
+ if err := provision.PutManifest(ctx, h.s, "bab", "app", provision.Manifest{Schema: 1, Moves: &provision.MovesManifest{ReadyPath: "/health"}}); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ if got := h.probePaths[len(h.probePaths)-1]; got != "/health" {
+ t.Fatalf("probed %q, want the declared /health", got)
+ }
+ // The manifest no longer parses: S1 fails, and "/" is never asked.
+ _, etag, err := h.s.Get(ctx, provision.ManifestKey("bab", "app"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := h.s.Put(ctx, provision.ManifestKey("bab", "app"), []byte("{not json"), etag); err != nil {
+ t.Fatal(err)
+ }
+ asked := len(h.probePaths)
+ h.pass(t)
+ if sr := h.sourceReport(t); sr.S1.OK || sr.S1.Reason != s1NoReadyPath {
+ t.Fatalf("S1 with an unreadable manifest: %+v", sr.S1)
+ }
+ if len(h.probePaths) != asked {
+ t.Fatalf("the instance was probed at %v with no readable manifest", h.probePaths[asked:])
+ }
+}
diff --git a/internal/host/movesprobe.go b/internal/host/movesprobe.go
new file mode 100644
index 000000000..49f6a3b49
--- /dev/null
+++ b/internal/host/movesprobe.go
@@ -0,0 +1,133 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "net"
+ "net/http"
+ "net/url"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// S1 (moves spec §7.3, codex 14; Task 14 review finding 4): the instance as
+// it runs today answers its readiness path from its own process. The probe
+// wakes a hibernating instance through the activator (ServeScheduled: the
+// same wake, no activity credit) and then asks the instance itself over its
+// own socket. Only that answer counts: anything the activator, the edge, a
+// gate or a fallback writes in its place is not the instance answering.
+
+// s1Timeout bounds one probe, wake included (§7.7's 30 s readiness window).
+const s1Timeout = 30 * time.Second
+
+// s1BodyLimit is the most of a readiness answer the probe reads; a longer
+// body fails, since it was not read in full.
+const s1BodyLimit = 1 << 20
+
+// probeInstance is S1 for rt at path. wake is the activator's entry
+// (Service.ServeScheduled in production): it wakes the instance and hands
+// the request to the handler it is given, which asks the instance over
+// rt.Socket. A wake that answers itself (a fenced or refused host, a
+// fallback) never reaches that handler. A route the activator does not
+// manage (wake returns false) is asked directly. S1 passes only on a 2xx
+// from the instance, its body read in full, within the deadline.
+func probeInstance(ctx context.Context, wake func(http.ResponseWriter, *http.Request, registry.Route, http.Handler) bool, rt registry.Route, path string) moves.Check {
+ fail := moves.Check{Reason: s1NoAnswer(rt.Host)}
+ ctx, cancel := context.WithTimeout(ctx, s1Timeout)
+ defer cancel()
+ if !strings.HasPrefix(path, "/") {
+ path = "/" + path
+ }
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+rt.Host+path, nil)
+ if err != nil {
+ return fail
+ }
+ req.Host = rt.Host
+ req.RemoteAddr = "127.0.0.1:0"
+ var answered bool
+ var status int
+ var askErr error
+ ask := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ answered = true
+ status, askErr = askInstance(r.Context(), rt.Socket, r)
+ if askErr == nil {
+ w.WriteHeader(status)
+ }
+ })
+ rec := &probeRecorder{header: http.Header{}}
+ if !wake(rec, req, rt, ask) {
+ ask.ServeHTTP(rec, req)
+ }
+ switch {
+ case !answered, askErr != nil, ctx.Err() != nil:
+ if askErr != nil {
+ log.Printf("moves: %s: S1: %v", rt.Host, askErr)
+ }
+ return fail
+ case status < 200 || status > 299:
+ return fail
+ }
+ return moves.Check{OK: true}
+}
+
+// askInstance sends r to the instance listening on socket and reads its
+// whole answer, never following a redirect.
+func askInstance(ctx context.Context, socket string, r *http.Request) (int, error) {
+ if socket == "" {
+ return 0, errors.New("the route names no socket")
+ }
+ client := &http.Client{
+ Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ var d net.Dialer
+ return d.DialContext(ctx, "unix", socket)
+ },
+ DisableKeepAlives: true,
+ },
+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+ }
+ out, err := http.NewRequestWithContext(ctx, http.MethodGet, (&url.URL{Scheme: "http", Host: r.Host, Path: r.URL.Path, RawQuery: r.URL.RawQuery}).String(), nil)
+ if err != nil {
+ return 0, err
+ }
+ out.Host = r.Host
+ resp, err := client.Do(out)
+ if err != nil {
+ return 0, err
+ }
+ defer resp.Body.Close()
+ n, err := io.Copy(io.Discard, io.LimitReader(resp.Body, s1BodyLimit+1))
+ if err != nil {
+ return 0, err
+ }
+ if n > s1BodyLimit {
+ return 0, fmt.Errorf("the answer is over %d bytes", s1BodyLimit)
+ }
+ return resp.StatusCode, nil
+}
+
+// probeRecorder is the ResponseWriter the wake writes to. What it records
+// is never evidence: only askInstance's answer is.
+type probeRecorder struct {
+ header http.Header
+ status int
+}
+
+func (r *probeRecorder) Header() http.Header { return r.header }
+func (r *probeRecorder) Write(b []byte) (int, error) {
+ if r.status == 0 {
+ r.status = http.StatusOK
+ }
+ return len(b), nil
+}
+func (r *probeRecorder) WriteHeader(code int) {
+ if r.status == 0 {
+ r.status = code
+ }
+}
diff --git a/internal/host/movesprobe_test.go b/internal/host/movesprobe_test.go
new file mode 100644
index 000000000..7dd7b8012
--- /dev/null
+++ b/internal/host/movesprobe_test.go
@@ -0,0 +1,126 @@
+package host
+
+import (
+ "context"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// instanceForTest serves h on a unix socket, the way an exec instance
+// listens, and returns the route that names it.
+func instanceForTest(t *testing.T, h http.HandlerFunc) registry.Route {
+ t.Helper()
+ dir, err := os.MkdirTemp("", "s1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ sock := filepath.Join(dir, "i.sock")
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srv := &http.Server{Handler: h}
+ go srv.Serve(ln)
+ t.Cleanup(func() { srv.Close() })
+ return registry.Route{Host: p1Host, Kind: "instance", Hibernate: true, Status: "ready", Socket: sock}
+}
+
+// wakeThenProxy is an activator that wakes and hands the request to the
+// probe's proxy, as Service.ServeScheduled does for an awake instance.
+func wakeThenProxy(w http.ResponseWriter, r *http.Request, _ registry.Route, proxy http.Handler) bool {
+ proxy.ServeHTTP(w, r)
+ return true
+}
+
+// S1 must prove the instance itself answered, in full, with a 2xx, within
+// its deadline (Task 14 review finding 4).
+func TestS1ProbeNeedsTheInstanceItself(t *testing.T) {
+ ctx := context.Background()
+ answer := func(code int) registry.Route {
+ return instanceForTest(t, func(w http.ResponseWriter, r *http.Request) {
+ if r.Host != p1Host || r.URL.Path != "/ready" {
+ w.WriteHeader(http.StatusTeapot)
+ return
+ }
+ w.WriteHeader(code)
+ w.Write([]byte("ok"))
+ })
+ }
+ if c := probeInstance(ctx, wakeThenProxy, answer(200), "/ready"); !c.OK {
+ t.Fatalf("a 200 from the instance: %+v", c)
+ }
+ if c := probeInstance(ctx, wakeThenProxy, answer(204), "/ready"); !c.OK {
+ t.Fatalf("a 204 from the instance: %+v", c)
+ }
+ for name, code := range map[string]int{"redirect": 302, "password gate": 401, "server error": 503, "not found": 404} {
+ if c := probeInstance(ctx, wakeThenProxy, answer(code), "/ready"); c.OK || c.Reason != s1NoAnswer(p1Host) {
+ t.Errorf("%s (%d) from the instance passed S1: %+v", name, code, c)
+ }
+ }
+
+ // Answers that never reached the instance: the edge's unknown-host
+ // fallback redirect, a gate's 401 and even a 200 written by the
+ // activator itself (a fenced or refused wake) all fail.
+ rt := answer(200)
+ for name, wake := range map[string]func(http.ResponseWriter, *http.Request, registry.Route, http.Handler) bool{
+ "fallback redirect": func(w http.ResponseWriter, r *http.Request, _ registry.Route, _ http.Handler) bool {
+ http.Redirect(w, r, "https://example.com/", http.StatusFound)
+ return true
+ },
+ "password gate": func(w http.ResponseWriter, _ *http.Request, _ registry.Route, _ http.Handler) bool {
+ w.WriteHeader(http.StatusUnauthorized)
+ return true
+ },
+ "a 200 not from the instance": func(w http.ResponseWriter, _ *http.Request, _ registry.Route, _ http.Handler) bool {
+ w.WriteHeader(http.StatusOK)
+ return true
+ },
+ } {
+ if c := probeInstance(ctx, wake, rt, "/ready"); c.OK {
+ t.Errorf("%s passed S1", name)
+ }
+ }
+
+ // 200 headers, part of a body, then nothing until the deadline.
+ slow := instanceForTest(t, func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Length", "100")
+ w.WriteHeader(http.StatusOK)
+ w.Write([]byte("partial"))
+ w.(http.Flusher).Flush()
+ select {
+ case <-r.Context().Done():
+ case <-time.After(5 * time.Second):
+ }
+ })
+ short, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
+ defer cancel()
+ if c := probeInstance(short, wakeThenProxy, slow, "/ready"); c.OK {
+ t.Fatal("a partial 200 that ran out its deadline passed S1")
+ }
+
+ // The instance answered in time, but the wake held the probe past its
+ // deadline: the probe expired, so S1 fails.
+ late, cancelLate := context.WithTimeout(ctx, 200*time.Millisecond)
+ defer cancelLate()
+ overrun := func(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
+ proxy.ServeHTTP(w, r)
+ <-r.Context().Done()
+ return true
+ }
+ if c := probeInstance(late, overrun, answer(200), "/ready"); c.OK {
+ t.Fatal("a probe that ran out its deadline passed S1")
+ }
+
+ // A route the activator does not manage is asked directly.
+ notManaged := func(http.ResponseWriter, *http.Request, registry.Route, http.Handler) bool { return false }
+ if c := probeInstance(ctx, notManaged, answer(200), "/ready"); !c.OK {
+ t.Fatalf("an always-on instance: %+v", c)
+ }
+}
diff --git a/internal/host/movesreport.go b/internal/host/movesreport.go
index 9e79b6fe1..20606fc52 100644
--- a/internal/host/movesreport.go
+++ b/internal/host/movesreport.go
@@ -8,26 +8,19 @@ import (
"fmt"
"io"
"log"
- "net/http"
"os"
"path/filepath"
- "sort"
- "strconv"
"strings"
- "sync/atomic"
"time"
- "github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/appconfig"
- "github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/moves/archive"
- "github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/provision"
- "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
)
// The target's readiness report (moves spec §7.3, T1 to T12), behind
@@ -59,6 +52,7 @@ const (
t10NoBucket = "the source region publishes no data bucket"
t10Failed = "this box cannot write the return lane in the source's bucket"
t11NoReplica = "no replica has been staged yet"
+ t11Unmeasured = "the database size could not be measured" // MV-N69
)
func t1Below(min int) string {
@@ -125,11 +119,14 @@ func loadMoveFleet(ctx context.Context, s release.Store) (moveFleet, error) {
// targetReportPass writes the target report for every move into this box's
// placement whose claim names this box as the target.
func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, recs []provision.Record, fleet moveFleet) {
+ live := map[string]bool{}
+ defer w.evictArchives(live)
for _, rec := range recs {
m := rec.Move
if m == nil || m.Legacy() || m.To != w.cfg.Region || (m.State != provision.MovePreparing && m.State != provision.MoveReady) {
continue
}
+ live[m.ID] = true
if err := w.targetReport(ctx, s, rec, fleet); err != nil {
w.logTrouble(rec.Host+" (target)", err)
continue
@@ -138,6 +135,18 @@ func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, rec
}
}
+// evictArchives drops T5's verdicts (and the carried .env they hold) for
+// every move no longer preparing or ready into this box.
+func (w *movesWorker) evictArchives(live map[string]bool) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ for id := range w.archives {
+ if !live[id] {
+ delete(w.archives, id)
+ }
+ }
+}
+
// verifiedArchive is what T5 learned from one published archive, kept per
// exact version so a pass does not open the same archive again.
type verifiedArchive struct {
@@ -188,8 +197,13 @@ func (w *movesWorker) targetReport(ctx context.Context, s release.Store, rec pro
rep.Checks[moves.CheckEnv] = t3
rep.Notes = notes
rep.Checks[moves.CheckStorage] = w.checkStorage(ctx, held, id)
- replicaBytes, replicaErr := w.stagedReplicaBytes(ctx, rec)
- rep.Checks[moves.CheckDisk] = w.checkDisk(replicaBytes, replicaErr, art)
+ // The database's logical size: the newest staged LTX header's, or the
+ // source's local file when that is larger (review finding 1).
+ staged, dbBytes, replicaErr := w.stagedReplica(ctx, rec)
+ if source != nil && source.DBBytes > dbBytes {
+ dbBytes = source.DBBytes
+ }
+ rep.Checks[moves.CheckDisk] = w.checkDisk(max(staged, dbBytes), replicaErr, art)
rep.Checks[moves.CheckRelease] = w.checkRelease(ctx, s, rec, source)
rep.Checks[moves.CheckOneOwner] = checkOneOwner(fleet, rec.Move.To)
rep.Checks[moves.CheckClaim] = w.checkClaim(ctx, obs, rec)
@@ -199,9 +213,9 @@ func (w *movesWorker) targetReport(ctx context.Context, s release.Store, rec pro
// T11 from measured rates: this box's own, and the source's cross rate.
m := w.measure(ctx, rec)
m.Cross = source.CrossMeasurement()
- dbMiB := float64(replicaBytes) / (1 << 20)
- if replicaErr != nil || replicaBytes == 0 {
- rep.Checks[moves.CheckBudget] = moves.Check{Reason: t11NoReplica}
+ dbMiB := float64(dbBytes) / (1 << 20)
+ if replicaErr != nil {
+ rep.Checks[moves.CheckBudget] = moves.Check{Reason: replicaReason(replicaErr)}
} else {
res := moves.BudgetCheck(dbMiB, art.entries, m, cfg.ReadyDelta, moves.CutWindowDefault, now)
rep.Checks[moves.CheckBudget] = res.Check
@@ -459,24 +473,105 @@ func (w *movesWorker) checkStorage(ctx context.Context, held heldEnv, id string)
return moves.Check{OK: true}
}
-// stagedReplicaBytes sums the replica staged in this box's bucket for the
-// move. It prices T6 and stands in for the database size in T11: the
-// staged LTX files hold at least the database's pages, so the estimate errs
-// long, never short.
-func (w *movesWorker) stagedReplicaBytes(ctx context.Context, rec provision.Record) (int64, error) {
+// errNoReplicaStaged is a lane with no LTX file in it yet.
+var errNoReplicaStaged = errors.New("no replica has been staged")
+
+func replicaReason(err error) string {
+ if errors.Is(err, errNoReplicaStaged) {
+ return t11NoReplica
+ }
+ return t11Unmeasured
+}
+
+// stagedReplica reads the replica staged in this box's bucket for the move:
+// the bytes staged, and the database's logical size, Commit pages of
+// PageSize from the header of the newest LTX file (the highest max TXID).
+// The staged bytes are no measure of the database: LTX compresses its
+// pages, so a compressible database stages far fewer bytes than it holds.
+// A newest file whose header cannot be read is an error, never a guess.
+func (w *movesWorker) stagedReplica(ctx context.Context, rec provision.Record) (staged, logical int64, err error) {
m, err := w.mirror(ctx, w.cfg.OwnBucketRegion)
if err != nil {
- return 0, err
+ return 0, 0, err
+ }
+ prefix := moves.ReplicaPrefix(rec.Host, rec.Move.ID)
+ objs, err := m.List(ctx, w.cfg.OwnBucket, prefix)
+ if err != nil {
+ return 0, 0, err
+ }
+ var newest string
+ var newestTX ltx.TXID
+ for k, o := range objs {
+ staged += o.Size
+ rel := strings.TrimPrefix(k, prefix)
+ _, name, ok := strings.Cut(rel, "/")
+ if !ok {
+ continue
+ }
+ _, maxTX, perr := ltx.ParseFilename(name)
+ if perr != nil {
+ continue
+ }
+ if newest == "" || maxTX > newestTX || (maxTX == newestTX && k > newest) {
+ newest, newestTX = k, maxTX
+ }
+ }
+ if newest == "" {
+ return staged, 0, errNoReplicaStaged
+ }
+ hdr, err := w.ltxHeader(ctx, newest)
+ if err != nil {
+ return staged, 0, fmt.Errorf("the newest staged LTX file %s: %w", newest, err)
+ }
+ return staged, int64(hdr.Commit) * int64(hdr.PageSize), nil
+}
+
+// ltxHeader reads the header of the current version of key in this box's
+// bucket: its first ltx.HeaderSize bytes, by exact version.
+func (w *movesWorker) ltxHeader(ctx context.Context, key string) (ltx.Header, error) {
+ cfg := w.cfg
+ if cfg.Versioned == nil {
+ return ltx.Header{}, errors.New("no versioned store")
+ }
+ vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+ if err != nil {
+ return ltx.Header{}, err
}
- objs, err := m.List(ctx, w.cfg.OwnBucket, moves.ReplicaPrefix(rec.Host, rec.Move.ID))
+ vers, err := vs.ListVersions(ctx, cfg.OwnBucket, key)
if err != nil {
- return 0, err
+ return ltx.Header{}, err
}
- var n int64
- for _, o := range objs {
- n += o.Size
+ var version string
+ for _, v := range vers {
+ if v.Key == key && v.IsLatest && !v.DeleteMarker {
+ version = v.VersionID
+ }
+ }
+ if version == "" {
+ return ltx.Header{}, errors.New("no current version")
+ }
+ var rc io.ReadCloser
+ if ro, ok := vs.(lease.VersionRangeOpener); ok {
+ rc, err = ro.OpenVersionRange(ctx, cfg.OwnBucket, key, version, 0, ltx.HeaderSize)
+ } else {
+ rc, err = vs.OpenVersion(ctx, cfg.OwnBucket, key, version)
+ }
+ if err != nil {
+ return ltx.Header{}, err
+ }
+ defer rc.Close()
+ buf := make([]byte, ltx.HeaderSize)
+ if _, err := io.ReadFull(rc, buf); err != nil {
+ return ltx.Header{}, err
}
- return n, nil
+ var hdr ltx.Header
+ if err := hdr.UnmarshalBinary(buf); err != nil {
+ return ltx.Header{}, err
+ }
+ if !ltx.IsValidPageSize(hdr.PageSize) || hdr.Commit == 0 {
+ return ltx.Header{}, fmt.Errorf("header names %d pages of %d bytes", hdr.Commit, hdr.PageSize)
+ }
+ return hdr, nil
}
// checkDisk is T6: free space covers twice the replica, plus the archive,
@@ -484,7 +579,7 @@ func (w *movesWorker) stagedReplicaBytes(ctx context.Context, rec provision.Reco
func (w *movesWorker) checkDisk(replicaBytes int64, replicaErr error, art verifiedArchive) moves.Check {
if replicaErr != nil {
log.Printf("moves: T6: %v", replicaErr)
- return moves.Check{Reason: t11NoReplica}
+ return moves.Check{Reason: replicaReason(replicaErr)}
}
if w.cfg.DiskFree == nil {
return moves.Check{Reason: t6Unmeasured}
@@ -622,272 +717,3 @@ func (w *movesWorker) measure(ctx context.Context, rec provision.Record) moves.M
w.mu.Unlock()
return m
}
-
-// ---- production hooks (armMovesPass wires them) ----
-
-// composeHeldEnv is T3's composition for a flat held row (ruling F19): a
-// scratch directory holding the configuration rendered from the merged
-// bundle (appconfig.LoadMergedHostVersions, rendered as RenderHeld renders
-// it) and the carried <host>.env, an ExecBackend over it with this box's
-// region and cache, and a synthetic flat route; LaunchEnvironment composes
-// what the next spawn would get.
-func composeHeldEnv(priv *ecdh.PrivateKey, region string, cache bool) func(context.Context, release.Store, provision.Record, []byte) (map[string]string, []string, provision.MoveInputs, error) {
- return func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, []string, provision.MoveInputs, error) {
- m, v, err := appconfig.LoadMergedHostVersions(ctx, s, rec.Account, rec.App, rec.Environment, rec.Host)
- if err != nil {
- return nil, nil, provision.MoveInputs{}, err
- }
- var rendered []byte
- if m != nil {
- if rendered, err = appconfig.RenderEnv(m, priv, appconfig.Binding{Account: rec.Account, App: rec.App}); err != nil {
- return nil, nil, provision.MoveInputs{}, err
- }
- }
- tmp, err := os.MkdirTemp("", "carlos-move-t3-")
- if err != nil {
- return nil, nil, provision.MoveInputs{}, err
- }
- defer os.RemoveAll(tmp)
- if err := os.WriteFile(filepath.Join(tmp, rec.Host+".config.env"), rendered, 0o600); err != nil {
- return nil, nil, provision.MoveInputs{}, err
- }
- if carried != nil {
- if err := os.WriteFile(filepath.Join(tmp, rec.Host+".env"), carried, 0o600); err != nil {
- return nil, nil, provision.MoveInputs{}, err
- }
- }
- eb := &activator.ExecBackend{DataDir: tmp, Region: region}
- if cache {
- eb.Cache = heldEnvCache
- }
- env, shadowed, err := eb.LaunchEnvironment(registry.Route{Host: rec.Host, Socket: provision.DefaultSocket(rec.Host)})
- if err != nil {
- return nil, nil, provision.MoveInputs{}, err
- }
- in := provision.MoveInputs{BundleETags: map[string]string{"default": v.Default, "environment": v.Environment}, OverlayETag: v.Overlay}
- return env, shadowed, in, nil
- }
-}
-
-// laneProbe is T10 in production: put-if-absent then delete-by-version of
-// moves/_lane/<id> in the source's bucket. A probe object left by an earlier
-// pass whose delete failed is removed by version and the check fails this
-// pass; the next one starts clean.
-func laneProbe(versioned func(ctx context.Context, bucket, region string) (lease.VersionedStore, error)) func(context.Context, string, string, string) error {
- return func(ctx context.Context, bucket, region, id string) error {
- vs, err := versioned(ctx, bucket, region)
- if err != nil {
- return err
- }
- key := moves.LaneKey(id)
- ver, err := vs.PutIfAbsent(ctx, bucket, key, []byte(`{"probe":"`+id+`"}`))
- if errors.Is(err, lease.ErrExists) {
- vers, lerr := vs.ListVersions(ctx, bucket, key)
- if lerr != nil {
- return fmt.Errorf("a probe object is left and cannot be listed: %w", lerr)
- }
- for _, v := range vers {
- if v.Key == key {
- if derr := vs.DeleteVersion(ctx, bucket, key, v.VersionID); derr != nil {
- return derr
- }
- }
- }
- return errors.New("a probe object was left by an earlier pass; removed")
- }
- if err != nil {
- return err
- }
- return vs.DeleteVersion(ctx, bucket, key, ver)
- }
-}
-
-// measureS3 times a 16 MiB put and get of the probe object (moves.ProbeKey,
-// outside moves/, F11) in this box's own bucket with its own credentials,
-// and deletes it.
-func measureS3(mirror func(ctx context.Context) (lease.MirrorStore, error), bucket string, now func() time.Time) func(context.Context, string) (moves.MeasuredRate, error) {
- return func(ctx context.Context, key string) (moves.MeasuredRate, error) {
- m, err := mirror(ctx)
- if err != nil {
- return moves.MeasuredRate{}, err
- }
- const size = 16 << 20
- body := make([]byte, size)
- start := now()
- if err := m.Put(ctx, bucket, key, body); err != nil {
- return moves.MeasuredRate{}, err
- }
- got, err := m.Get(ctx, bucket, key)
- elapsed := now().Sub(start)
- if derr := m.Delete(ctx, bucket, key); derr != nil {
- log.Printf("moves: delete the S3 probe object %s: %v", key, derr)
- }
- if err != nil {
- return moves.MeasuredRate{}, err
- }
- if len(got) != size {
- return moves.MeasuredRate{}, fmt.Errorf("the probe object read back %d bytes of %d", len(got), size)
- }
- return moves.RateOf(2*size, elapsed, now().UTC()), nil
- }
-}
-
-// measureDiskIn measures this box's disk under <dataDir>/.moves (the
-// platform-specific write and read are movesmeasure_*.go).
-func measureDiskIn(dataDir string, now func() time.Time) func(context.Context) (moves.MeasuredRate, moves.FilesRate, error) {
- return func(ctx context.Context) (moves.MeasuredRate, moves.FilesRate, error) {
- dir, err := os.MkdirTemp(filepath.Join(dataDir, ".moves"), "measure-")
- if err != nil {
- if err := os.MkdirAll(filepath.Join(dataDir, ".moves"), 0o755); err != nil {
- return moves.MeasuredRate{}, moves.FilesRate{}, err
- }
- if dir, err = os.MkdirTemp(filepath.Join(dataDir, ".moves"), "measure-"); err != nil {
- return moves.MeasuredRate{}, moves.FilesRate{}, err
- }
- }
- defer os.RemoveAll(dir)
- disk, err := measureDiskRate(dir, now)
- if err != nil {
- return moves.MeasuredRate{}, moves.FilesRate{}, err
- }
- files, err := measureFsyncRate(dir, now)
- if err != nil {
- return moves.MeasuredRate{}, moves.FilesRate{}, err
- }
- return disk, files, nil
- }
-}
-
-// measureFsyncFiles and measureFsyncWorkers are the per-file sync probe:
-// a few hundred small files synced by the extract's 16-worker pattern
-// (ruling T7-budget).
-const (
- measureFsyncFiles = 512
- measureFsyncWorkers = 16
-)
-
-func measureFsyncRate(dir string, now func() time.Time) (moves.FilesRate, error) {
- jobs := make(chan int)
- errs := make(chan error, measureFsyncWorkers)
- start := now()
- for i := 0; i < measureFsyncWorkers; i++ {
- go func() {
- var first error
- for n := range jobs {
- if first != nil {
- continue
- }
- f, err := os.OpenFile(filepath.Join(dir, "f"+strconv.Itoa(n)), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
- if err == nil {
- _, err = f.Write(make([]byte, 4096))
- if err == nil {
- err = f.Sync()
- }
- if cerr := f.Close(); err == nil {
- err = cerr
- }
- }
- first = err
- }
- errs <- first
- }()
- }
- for n := 0; n < measureFsyncFiles; n++ {
- jobs <- n
- }
- close(jobs)
- for i := 0; i < measureFsyncWorkers; i++ {
- if err := <-errs; err != nil {
- return moves.FilesRate{}, err
- }
- }
- d := now().Sub(start)
- if d <= 0 {
- d = time.Microsecond
- }
- return moves.FilesRate{PerS: measureFsyncFiles / d.Seconds(), At: now().UTC(), Files: measureFsyncFiles}, nil
-}
-
-// heldEnvCache is the non-nil cache T3's synthetic backend carries when the
-// live box has a response cache: LaunchEnvironment reads only whether one
-// is set (CARLOS_PURGE).
-var heldEnvCache = edge.NewCache(1, 1)
-
-// storePrefixes are the store bindings in an environment: every prefix P
-// with P_BUCKET, P_ACCESS_KEY and P_SECRET_KEY set (objstore.EnvNames).
-func storePrefixes(env map[string]string) []string {
- var out []string
- for name := range env {
- p, ok := strings.CutSuffix(name, "_BUCKET")
- if !ok || p == "" {
- continue
- }
- _, _, _, ak, sk := objstore.EnvNames(p)
- if env[ak] != "" && env[sk] != "" {
- out = append(out, p)
- }
- }
- sort.Strings(out)
- return out
-}
-
-// movesEdge is the agent's edge, once RunWithActivatorReady has one.
-var movesEdge atomic.Pointer[edge.Edge]
-
-// setMovesReadyProbe hands the worker's S1 probe the agent's edge
-// (agent.go, RunWithActivatorReady's callback).
-func setMovesReadyProbe(e *edge.Edge) { movesEdge.Store(e) }
-
-// probeThroughEdge is S1 (§7.3, round-4 finding 11): GET path on host
-// through the edge's uncached path, which wakes a hibernating source and
-// never answers from, or stores into, the response cache. 200 to 399 within
-// 30 s passes.
-func probeThroughEdge(ctx context.Context, host, path string) moves.Check {
- e := movesEdge.Load()
- if e == nil {
- return moves.Check{Reason: s1EdgeNotReady}
- }
- return probeWith(ctx, e.ServeUncached, host, path)
-}
-
-func probeWith(ctx context.Context, serve func(http.ResponseWriter, *http.Request), host, path string) moves.Check {
- ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
- defer cancel()
- if !strings.HasPrefix(path, "/") {
- path = "/" + path
- }
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+host+path, nil)
- if err != nil {
- return moves.Check{Reason: s1NoAnswer(host)}
- }
- req.Host = host
- req.RemoteAddr = "127.0.0.1:0"
- rec := &probeRecorder{header: http.Header{}}
- serve(rec, req)
- if rec.status == 0 {
- rec.status = http.StatusOK // a handler that wrote a body without a header
- }
- if rec.status < 200 || rec.status > 399 {
- return moves.Check{Reason: s1NoAnswer(host)}
- }
- return moves.Check{OK: true}
-}
-
-// probeRecorder keeps the status and discards the body.
-type probeRecorder struct {
- header http.Header
- status int
-}
-
-func (r *probeRecorder) Header() http.Header { return r.header }
-func (r *probeRecorder) Write(b []byte) (int, error) {
- if r.status == 0 {
- r.status = http.StatusOK
- }
- return len(b), nil
-}
-func (r *probeRecorder) WriteHeader(code int) {
- if r.status == 0 {
- r.status = code
- }
-}
diff --git a/internal/host/movesreport_minio_test.go b/internal/host/movesreport_minio_test.go
index c1ccfc0a5..3eae1514a 100644
--- a/internal/host/movesreport_minio_test.go
+++ b/internal/host/movesreport_minio_test.go
@@ -9,6 +9,7 @@ import (
"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"
+ s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/aws/smithy-go"
)
@@ -46,9 +47,12 @@ func TestT4StorageProbeMinio(t *testing.T) {
ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
})
_, err := c.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: aws.String(bucket), Key: aws.String(".carlos-move-probe/" + p1ID)})
+ // Only a not-found answer proves it is gone: AccessDenied or any other
+ // error says nothing about the object.
var apiErr smithy.APIError
- if err == nil || !errors.As(err, &apiErr) {
- t.Fatalf("the probe object was left behind (head: %v)", err)
+ var notFound *s3types.NotFound
+ if err == nil || !(errors.As(err, &notFound) || (errors.As(err, &apiErr) && (apiErr.ErrorCode() == "NotFound" || apiErr.ErrorCode() == "NoSuchKey"))) {
+ t.Fatalf("the probe object is not shown gone (head: %v)", err)
}
// Wrong credentials: T4 fails with the store's sentence.
diff --git a/internal/host/movesreport_test.go b/internal/host/movesreport_test.go
index e1e4c7218..cede77dc6 100644
--- a/internal/host/movesreport_test.go
+++ b/internal/host/movesreport_test.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "encoding/binary"
"encoding/json"
"errors"
"fmt"
@@ -321,29 +322,6 @@ func TestPreparingRelaysHeldStageOnAHardenedBox(t *testing.T) {
}
}
-// S1 through the edge's uncached path: 200 to 399 passes, anything else is
-// M-C16; before the edge exists, the probe says so.
-func TestReadyProbeThroughTheEdge(t *testing.T) {
- ctx := context.Background()
- for code, ok := range map[int]bool{200: true, 302: true, 399: true, 404: false, 503: false} {
- serve := func(w http.ResponseWriter, r *http.Request) {
- if r.Host != p1Host || r.URL.Path != "/ready" {
- w.WriteHeader(http.StatusTeapot)
- return
- }
- w.WriteHeader(code)
- }
- c := probeWith(ctx, serve, p1Host, "/ready")
- if c.OK != ok || (!ok && c.Reason != s1NoAnswer(p1Host)) {
- t.Errorf("status %d: %+v", code, c)
- }
- }
- movesEdge.Store(nil)
- if c := probeThroughEdge(ctx, p1Host, "/"); c.OK || c.Reason != s1EdgeNotReady {
- t.Fatalf("no edge: %+v", c)
- }
-}
-
// T4 finds the store bindings in the composed environment by objstore's
// variable names; a prefix without its credentials is not a binding.
func TestStorePrefixesFollowTheDeliveredNames(t *testing.T) {
@@ -525,3 +503,88 @@ func TestProbeStoreSpeaksS3(t *testing.T) {
t.Fatalf("wrong credentials: %q %v", prefix, err)
}
}
+
+// T11 prices the database's logical size, read from the newest staged LTX
+// header (Commit pages of PageSize), never the compressed bytes staged
+// (Task 14 review finding 1). The source's local database file counts when
+// it is larger; a newest file that is not LTX fails T6 and T11 closed.
+func TestT11PricesTheLogicalDatabaseSize(t *testing.T) {
+ h := newPhase1Harness(t)
+ h.start(t, "manual")
+ newest := ltxForTest(t, 4096, 1<<18, 3, 3) // a 1 GiB database, one page in this file
+ if len(newest) > 64<<10 {
+ t.Fatalf("the fixture is %d bytes: not compressed enough to prove anything", len(newest))
+ }
+ key := "instances/" + p1Host + "/" + h.inc + "/g-cur/0000/0000000000000003-0000000000000003.ltx"
+ h.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "newest", Size: int64(len(newest))}, string(newest)})
+ h.pass(t)
+ h.targetPass(t, h.targetCfg())
+ r := h.targetReport(t)
+ if r.DBMiB != 1024 {
+ t.Fatalf("T11 priced %v MiB, want the 1024 MiB the newest header names (staged bytes are about %d KiB)", r.DBMiB, len(newest)>>10)
+ }
+
+ // A larger local database file on the source wins.
+ db, err := os.Create(provision.DefaultDBPath(h.cfg.DataDir, p1Host))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := db.Truncate(2 << 30); err != nil { // sparse
+ t.Fatal(err)
+ }
+ db.Close()
+ h.pass(t)
+ if sr := h.sourceReport(t); sr.DBBytes != 2<<30 {
+ t.Fatalf("source DBBytes %d", sr.DBBytes)
+ }
+ h.targetPass(t, h.targetCfg())
+ if r := h.targetReport(t); r.DBMiB != 2048 {
+ t.Fatalf("T11 priced %v MiB with a 2 GiB source file", r.DBMiB)
+ }
+
+ // The newest staged file is not LTX: the size is unknown, so T6 and
+ // T11 fail closed.
+ h2 := newPhase1Harness(t)
+ h2.start(t, "manual")
+ h2.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "junk", Size: 4}, "junk"})
+ h2.pass(t)
+ h2.targetPass(t, h2.targetCfg())
+ f := failing(h2.targetReport(t))
+ if f[moves.CheckBudget] != t11Unmeasured || f[moves.CheckDisk] != t11Unmeasured {
+ t.Fatalf("an unreadable newest header: %v", f)
+ }
+
+ // A header that parses but names no pages is no measurement either.
+ zero := append([]byte(nil), newest...)
+ binary.BigEndian.PutUint32(zero[12:], 0) // Commit
+ h3 := newPhase1Harness(t)
+ h3.start(t, "manual")
+ h3.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "zero", Size: int64(len(zero))}, string(zero)})
+ h3.pass(t)
+ h3.targetPass(t, h3.targetCfg())
+ if f := failing(h3.targetReport(t)); f[moves.CheckBudget] != t11Unmeasured {
+ t.Fatalf("a header naming zero pages: %v", f)
+ }
+}
+
+// T5's verdicts (with the carried .env) are kept only while their move is
+// in flight into this box.
+func TestArchiveVerdictsAreEvictedWhenTheMoveEnds(t *testing.T) {
+ ctx := context.Background()
+ h := newPhase1Harness(t)
+ h.start(t, "manual")
+ h.pass(t)
+ h.targetPass(t, h.targetCfg())
+ if len(h.worker2.archives) != 1 {
+ t.Fatalf("%d verdicts cached", len(h.worker2.archives))
+ }
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t) // the abort completes; the record is at rest
+ h.targetPass(t, h.targetCfg())
+ if len(h.worker2.archives) != 0 {
+ t.Fatalf("a verdict outlived its move: %v", h.worker2.archives)
+ }
+}
diff --git a/internal/host/movesstage.go b/internal/host/movesstage.go
index e77a99abc..c51d72b60 100644
--- a/internal/host/movesstage.go
+++ b/internal/host/movesstage.go
@@ -1,88 +1,19 @@
package host
import (
- "bytes"
"context"
"encoding/json"
"errors"
- "fmt"
- "io"
"log"
"os"
"path/filepath"
- "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/moves"
- "github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
-// probeStores is T4 in production (§7.3): for each store bound in the
-// composed environment (objstore.EnvNames), with that store's own
-// credentials, list one key, then put, get and delete
-// .carlos-move-probe/<move id>. It returns the first failing binding's
-// prefix with its error. The client is built the way a member app builds
-// it from the delivered variables (objstore's delivered contract): static
-// credentials, the delivered region, and the endpoint path-style when one
-// is delivered.
-func probeStores(ctx context.Context, env map[string]string, id string) (string, error) {
- for _, p := range storePrefixes(env) {
- if err := probeStore(ctx, env, p, id); err != nil {
- return p, err
- }
- }
- return "", nil
-}
-
-func probeStore(ctx context.Context, env map[string]string, prefix, id string) error {
- bn, rn, en, ak, sk := objstore.EnvNames(prefix)
- region := env[rn]
- if region == "" {
- region = "us-east-1"
- }
- opts := awss3.Options{
- Region: region,
- Credentials: credentials.NewStaticCredentialsProvider(env[ak], env[sk], ""),
- RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
- ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
- }
- if ep := env[en]; ep != "" {
- opts.BaseEndpoint = aws.String(ep)
- opts.UsePathStyle = true
- }
- c := awss3.New(opts)
- bucket := aws.String(env[bn])
- key := aws.String(".carlos-move-probe/" + id)
- body := []byte("carlos move probe " + id)
- if _, err := c.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{Bucket: bucket, MaxKeys: aws.Int32(1)}); err != nil {
- return fmt.Errorf("list: %w", err)
- }
- if _, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: bucket, Key: key, Body: bytes.NewReader(body)}); err != nil {
- return fmt.Errorf("put: %w", err)
- }
- out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: bucket, Key: key})
- if err != nil {
- return fmt.Errorf("get: %w", err)
- }
- got, err := io.ReadAll(io.LimitReader(out.Body, 1<<16))
- out.Body.Close()
- if err != nil {
- return fmt.Errorf("get: %w", err)
- }
- if !bytes.Equal(got, body) {
- return errors.New("get: the probe object read back different bytes")
- }
- if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: bucket, Key: key}); err != nil {
- return fmt.Errorf("delete: %w", err)
- }
- return nil
-}
-
// heldMaterialize is release.MaterializeHeld as a variable, so a test can
// make the stage fail the way a read-only version store does on a hardened
// box (the sandboxed agent cannot write /opt/carlos).
diff --git a/internal/host/movesstore.go b/internal/host/movesstore.go
new file mode 100644
index 000000000..7d1b1e9e1
--- /dev/null
+++ b/internal/host/movesstore.go
@@ -0,0 +1,135 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "sort"
+ "strings"
+
+ "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/moves"
+ "github.com/carlosframework/platform/internal/objstore"
+)
+
+// The storage probes of the target's report: T4's probe of every store
+// bound in the composed environment, and T10's return lane in the source's
+// bucket. armPhase1 wires them as moveCfg.ProbeStore and LaneProbe.
+
+// probeStores is T4 in production (§7.3): for each store bound in the
+// composed environment (objstore.EnvNames), with that store's own
+// credentials, list one key, then put, get and delete
+// .carlos-move-probe/<move id>. It returns the first failing binding's
+// prefix with its error. The client is built the way a member app builds
+// it from the delivered variables (objstore's delivered contract): static
+// credentials, the delivered region, and the endpoint path-style when one
+// is delivered.
+func probeStores(ctx context.Context, env map[string]string, id string) (string, error) {
+ for _, p := range storePrefixes(env) {
+ if err := probeStore(ctx, env, p, id); err != nil {
+ return p, err
+ }
+ }
+ return "", nil
+}
+
+func probeStore(ctx context.Context, env map[string]string, prefix, id string) error {
+ bn, rn, en, ak, sk := objstore.EnvNames(prefix)
+ region := env[rn]
+ if region == "" {
+ region = "us-east-1"
+ }
+ opts := awss3.Options{
+ Region: region,
+ Credentials: credentials.NewStaticCredentialsProvider(env[ak], env[sk], ""),
+ RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+ ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
+ }
+ if ep := env[en]; ep != "" {
+ opts.BaseEndpoint = aws.String(ep)
+ opts.UsePathStyle = true
+ }
+ c := awss3.New(opts)
+ bucket := aws.String(env[bn])
+ key := aws.String(".carlos-move-probe/" + id)
+ body := []byte("carlos move probe " + id)
+ if _, err := c.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{Bucket: bucket, MaxKeys: aws.Int32(1)}); err != nil {
+ return fmt.Errorf("list: %w", err)
+ }
+ if _, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: bucket, Key: key, Body: bytes.NewReader(body)}); err != nil {
+ return fmt.Errorf("put: %w", err)
+ }
+ out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: bucket, Key: key})
+ if err != nil {
+ return fmt.Errorf("get: %w", err)
+ }
+ got, err := io.ReadAll(io.LimitReader(out.Body, 1<<16))
+ out.Body.Close()
+ if err != nil {
+ return fmt.Errorf("get: %w", err)
+ }
+ if !bytes.Equal(got, body) {
+ return errors.New("get: the probe object read back different bytes")
+ }
+ if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: bucket, Key: key}); err != nil {
+ return fmt.Errorf("delete: %w", err)
+ }
+ return nil
+}
+
+// laneProbe is T10 in production: put-if-absent then delete-by-version of
+// moves/_lane/<id> in the source's bucket. A probe object left by an earlier
+// pass whose delete failed is removed by version and the check fails this
+// pass; the next one starts clean.
+func laneProbe(versioned func(ctx context.Context, bucket, region string) (lease.VersionedStore, error)) func(context.Context, string, string, string) error {
+ return func(ctx context.Context, bucket, region, id string) error {
+ vs, err := versioned(ctx, bucket, region)
+ if err != nil {
+ return err
+ }
+ key := moves.LaneKey(id)
+ ver, err := vs.PutIfAbsent(ctx, bucket, key, []byte(`{"probe":"`+id+`"}`))
+ if errors.Is(err, lease.ErrExists) {
+ vers, lerr := vs.ListVersions(ctx, bucket, key)
+ if lerr != nil {
+ return fmt.Errorf("a probe object is left and cannot be listed: %w", lerr)
+ }
+ for _, v := range vers {
+ if v.Key == key {
+ if derr := vs.DeleteVersion(ctx, bucket, key, v.VersionID); derr != nil {
+ return derr
+ }
+ }
+ }
+ return errors.New("a probe object was left by an earlier pass; removed")
+ }
+ if err != nil {
+ return err
+ }
+ return vs.DeleteVersion(ctx, bucket, key, ver)
+ }
+}
+
+// storePrefixes are the store bindings in an environment: every prefix P
+// with P_BUCKET, P_ACCESS_KEY and P_SECRET_KEY set (objstore.EnvNames).
+func storePrefixes(env map[string]string) []string {
+ var out []string
+ for name := range env {
+ p, ok := strings.CutSuffix(name, "_BUCKET")
+ if !ok || p == "" {
+ continue
+ }
+ _, _, _, ak, sk := objstore.EnvNames(p)
+ if env[ak] != "" && env[sk] != "" {
+ out = append(out, p)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
diff --git a/internal/lease/staging.go b/internal/lease/staging.go
index d31e37cb9..4152afbab 100644
--- a/internal/lease/staging.go
+++ b/internal/lease/staging.go
@@ -54,6 +54,9 @@ type StagingWriter interface {
CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (etag, versionID string, err error)
// PutVersionedAt writes body over key and returns the new version id.
PutVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (versionID string, err error)
+ // PutReaderVersionedAt is PutVersionedAt streaming size bytes from r,
+ // signed with the sha256 the caller computed while writing them.
+ PutReaderVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, r io.Reader, size int64, sha256Hex string) (versionID string, err error)
// PutIfAbsentVersionedAt is PutVersionedAt with If-None-Match: *; an
// existing object gives ErrExists.
PutIfAbsentVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (versionID string, err error)
@@ -228,6 +231,18 @@ func (m *StagingMirror) objectURL(bucket, key string) (*url.URL, error) {
// clock, retries, or re-signs. It returns the status, headers and (capped)
// body of whatever S3 answered.
func (m *StagingMirror) send(ctx context.Context, signAt time.Time, method, bucket, key string, hdr http.Header, body []byte) (int, http.Header, []byte, error) {
+ if body == nil {
+ return m.sendBody(ctx, signAt, method, bucket, key, hdr, nil, 0, unsignedPayload)
+ }
+ sum := sha256.Sum256(body)
+ return m.sendBody(ctx, signAt, method, bucket, key, hdr, bytes.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:]))
+}
+
+// sendBody is send with the body as a reader of a declared size whose
+// sha256 (hex) the caller already knows: it is the signed payload hash, so
+// S3 refuses a body that does not match it, and a reader that ends short of
+// size fails the request rather than landing a truncated object.
+func (m *StagingMirror) sendBody(ctx context.Context, signAt time.Time, method, bucket, key string, hdr http.Header, rd io.Reader, size int64, payloadHash string) (int, http.Header, []byte, error) {
if signAt.IsZero() {
return 0, nil, nil, errors.New("staging mirror: no signing time (a mutation signs at a Guarded.Before check)")
}
@@ -235,17 +250,13 @@ func (m *StagingMirror) send(ctx context.Context, signAt time.Time, method, buck
if err != nil {
return 0, nil, nil, err
}
- var rd io.Reader
- payloadHash := unsignedPayload
- if body != nil {
- rd = bytes.NewReader(body)
- sum := sha256.Sum256(body)
- payloadHash = hex.EncodeToString(sum[:])
- }
req, err := http.NewRequestWithContext(ctx, method, u.String(), rd)
if err != nil {
return 0, nil, nil, err
}
+ if rd != nil {
+ req.ContentLength = size
+ }
for k, vs := range hdr {
req.Header[k] = vs
}
@@ -356,6 +367,28 @@ func (m *StagingMirror) PutVersionedAt(ctx context.Context, signAt time.Time, bu
return m.put(ctx, signAt, bucket, key, body, false)
}
+// PutReaderVersionedAt is PutVersionedAt for a body read from r: size bytes
+// whose sha256 is sha256Hex. The archive is staged this way from its file,
+// so a 256 MiB archive is never held in memory.
+func (m *StagingMirror) PutReaderVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, r io.Reader, size int64, sha256Hex string) (string, error) {
+ if size < 0 {
+ return "", fmt.Errorf("staging mirror: put %s/%s: negative size", bucket, key)
+ }
+ if b, err := hex.DecodeString(sha256Hex); err != nil || len(b) != sha256.Size {
+ return "", fmt.Errorf("staging mirror: put %s/%s: not a sha256", bucket, key)
+ }
+ hdr := http.Header{}
+ hdr.Set("Content-Type", "application/octet-stream")
+ status, h, rb, err := m.sendBody(ctx, signAt, http.MethodPut, bucket, key, hdr, io.LimitReader(r, size), size, sha256Hex)
+ if err != nil {
+ return "", err
+ }
+ if status/100 != 2 {
+ return "", stagingFailure("put "+bucket+"/"+key, status, rb)
+ }
+ return versionIDOf("put "+bucket+"/"+key, h)
+}
+
func (m *StagingMirror) PutIfAbsentVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) {
return m.put(ctx, signAt, bucket, key, body, true)
}
diff --git a/internal/lease/staging_test.go b/internal/lease/staging_test.go
index 46204e45e..e67800778 100644
--- a/internal/lease/staging_test.go
+++ b/internal/lease/staging_test.go
@@ -374,3 +374,48 @@ func TestStagingMirrorNeverFollowsARedirect(t *testing.T) {
}
}
}
+
+// PutReaderVersionedAt streams a body of a known size and sha256 (the
+// sealed archive, kept on disk) under the same single signature: the
+// payload hash it signs is the one given, the body lands whole, and a
+// reader that ends early is never sent as a complete object.
+func TestStagingMirrorPutReader(t *testing.T) {
+ m, bucket, reqs := stagingFixture(t, nil)
+ ctx := context.Background()
+ body := strings.Repeat("sealed ", 1000)
+ sum := sha256.Sum256([]byte(body))
+ ver, err := m.PutReaderVersionedAt(ctx, signAt, "data-uk", "moves/h/op/files/files-x.sealed", strings.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:]))
+ if err != nil || ver == "" {
+ t.Fatalf("put: %q %v", ver, err)
+ }
+ r := reqs()[0]
+ if r.sigErr != nil || r.payload != hex.EncodeToString(sum[:]) || r.body != body || r.ifNoneMatch != "" {
+ t.Fatalf("request %+v", r.sigErr)
+ }
+ if got, ok := bucket.Current("data-uk", "moves/h/op/files/files-x.sealed"); !ok || string(got) != body {
+ t.Fatalf("not stored: %q", got)
+ }
+ // A hash that is not 64 hex digits, a negative size, or a zero signing
+ // time sends nothing.
+ n := len(reqs())
+ for _, c := range []struct {
+ size int64
+ sha string
+ at time.Time
+ }{{int64(len(body)), "nothex", signAt}, {-1, hex.EncodeToString(sum[:]), signAt}, {int64(len(body)), hex.EncodeToString(sum[:]), time.Time{}}} {
+ if _, err := m.PutReaderVersionedAt(ctx, c.at, "data-uk", "k", strings.NewReader(body), c.size, c.sha); err == nil {
+ t.Fatalf("accepted size %d sha %q at %v", c.size, c.sha, c.at)
+ }
+ }
+ if len(reqs()) != n {
+ t.Fatal("a refused put sent a request")
+ }
+ // A reader shorter than its declared size fails rather than landing a
+ // truncated object.
+ if _, err := m.PutReaderVersionedAt(ctx, signAt, "data-uk", "short", strings.NewReader("abc"), 10, hex.EncodeToString(sum[:])); err == nil {
+ t.Fatal("a short body was accepted")
+ }
+ if _, ok := bucket.Current("data-uk", "short"); ok {
+ t.Fatal("a short body landed")
+ }
+}
diff --git a/internal/moves/readiness.go b/internal/moves/readiness.go
index 000ffdecd..b804597bd 100644
--- a/internal/moves/readiness.go
+++ b/internal/moves/readiness.go
@@ -99,15 +99,19 @@ func (r *TargetReport) RecordBudget(dbMiB float64, entries int, m Measurements,
// the source's most recent copying round, made through the target bucket's
// region (the path the cut uses), and CrossRateAt is when that round ended.
type SourceReport struct {
- V int `json:"v"`
- Box string `json:"box"`
- MoveID string `json:"move_id"`
- At time.Time `json:"at"`
- Env EnvReport `json:"env"`
- S1 Check `json:"s1"`
- S2 Check `json:"s2"`
- FilesPlain string `json:"files_plain"`
- Release string `json:"release"`
+ V int `json:"v"`
+ Box string `json:"box"`
+ MoveID string `json:"move_id"`
+ At time.Time `json:"at"`
+ Env EnvReport `json:"env"`
+ S1 Check `json:"s1"`
+ S2 Check `json:"s2"`
+ FilesPlain string `json:"files_plain"`
+ Release string `json:"release"`
+ // DBBytes is the size of the source's local database file when one is
+ // present (a hibernating source may have none). T11 prices the larger
+ // of it and the logical size in the newest staged LTX header.
+ DBBytes int64 `json:"db_bytes,omitempty"`
Archive *ArchiveRef `json:"archive,omitempty"`
CrossRate float64 `json:"cross_rate"`
CrossRateAt time.Time `json:"cross_rate_at"`
@@ -162,7 +166,7 @@ func (r TargetReport) Passing() (bool, string) {
// move back out of ready (Review Focus 5).
//
// The target's checks must also have been made against what the source
-// publishes now (round-5 finding 9): a fresh passing report for archive A
+// publishes now (round-5 finding 9), its release included: a fresh passing report for archive A
// must not promote after the source has published archive B. And the budget
// must have been priced for a final mirror round as large as the bar allows
// (F28(e)), so raising CARLOS_MOVE_READY_DELTA_BYTES cannot promote on an
@@ -215,6 +219,14 @@ func Ready(roundDelta, bar int64, target *TargetReport, source *SourceReport, no
if target.Inputs.EnvReport != source.Env.Digest {
return false, fmt.Sprintf("the target checked launch environment %s and the source now reports %s", target.Inputs.EnvReport, source.Env.Digest)
}
+ // The release promotion pins is the one T7 established on the target,
+ // never one the source moved to since (Task 14 review finding 3).
+ if target.Inputs.Release == "" {
+ return false, "the target has not checked a release yet"
+ }
+ if target.Inputs.Release != source.Release {
+ return false, fmt.Sprintf("the target checked release %s and the source now runs %s", target.Inputs.Release, source.Release)
+ }
return true, ""
}
diff --git a/internal/moves/readiness_test.go b/internal/moves/readiness_test.go
index 2a6ba1fcf..1af96f089 100644
--- a/internal/moves/readiness_test.go
+++ b/internal/moves/readiness_test.go
@@ -21,7 +21,7 @@ import (
// 100 MiB database priced from rates measured a minute earlier, in a 120 s
// window. goodSource is the matching source report.
func passingTarget(at time.Time) *TargetReport {
- r := &TargetReport{V: 1, At: at, Checks: map[string]Check{}, CutBudgetS: 120}
+ r := &TargetReport{V: 1, At: at, Checks: map[string]Check{}, CutBudgetS: 120, Inputs: provision.MoveInputs{Release: "v7"}}
for _, id := range AllChecks {
r.Checks[id] = Check{OK: true}
}
@@ -46,7 +46,7 @@ func sourceWith(now time.Time, s1, s2 Check) *SourceReport {
}
func goodSource(now time.Time) *SourceReport {
- return &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}, CrossRate: 10, CrossRateAt: now.Add(-time.Minute)}
+ return &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}, CrossRate: 10, CrossRateAt: now.Add(-time.Minute), Release: "v7"}
}
func TestReadyNeedsFreshReports(t *testing.T) {
@@ -138,7 +138,7 @@ func TestReadyRejectsMismatchedInputs(t *testing.T) {
now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
mk := func(targetFiles, sourceFiles, targetEnv, sourceEnv string) (*TargetReport, *SourceReport) {
tr := passingTarget(now)
- tr.Inputs = provision.MoveInputs{FilesPlain: targetFiles, EnvReport: targetEnv}
+ tr.Inputs = provision.MoveInputs{FilesPlain: targetFiles, EnvReport: targetEnv, Release: "v7"}
sr := goodSource(now)
sr.FilesPlain, sr.Env = sourceFiles, EnvReport{V: 1, Digest: sourceEnv}
return tr, sr
@@ -799,3 +799,29 @@ func TestT11BudgetCheckPricesEntries(t *testing.T) {
t.Fatalf("RecordBudget dropped the entry evidence: %+v", tr)
}
}
+
+// The release a promotion pins is one the target checked (T7): a target
+// report with no release, or one for another release than the source now
+// runs, never promotes (Task 14 review finding 3).
+func TestReadyNeedsTheReleaseTheTargetChecked(t *testing.T) {
+ now := t11Now
+ if ok, why := Ready(0, 16<<20, passingTarget(now), goodSource(now), now); !ok {
+ t.Fatal(why)
+ }
+ tr := passingTarget(now)
+ tr.Inputs.Release = ""
+ if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); ok || why != "the target has not checked a release yet" {
+ t.Fatalf("no release checked: %v %q", ok, why)
+ }
+ src := goodSource(now)
+ src.Release = "v8"
+ if ok, why := Ready(0, 16<<20, passingTarget(now), src, now); ok || why != "the target checked release v7 and the source now runs v8" {
+ t.Fatalf("release drift: %v %q", ok, why)
+ }
+ src.Release = ""
+ tr = passingTarget(now)
+ tr.Inputs.Release = ""
+ if ok, _ := Ready(0, 16<<20, tr, src, now); ok {
+ t.Fatal("two empty releases promoted")
+ }
+}