amadan

carlos / platform Public

activator, hostbox, lease, schedule: the moves interfaces and cold admitted-start sequence (claim Plan 6 Task 6)

startAdmitted with startPolicy{MaxStarts, Window, Deadline, Probe,
Prepared, Inputs}, K6 in the wake admission, SelectedTimers for held rows
with ReplaceTimers, generation-bound replication (no bare-key LTX
deletes), cold restore through genselect; inert unless CARLOS_HOST_CLAIM.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude bf63e2506dac4175543d4e99779c1a4ca3681f4d parent 2bd3ea1
19 files changed, +1915 −12
  • internal/activator/activator.go +11 −4
  • internal/activator/claimspawn.go +12 −0
  • internal/activator/claimwake.go +334 −0
  • internal/activator/claimwake_test.go +75 −0
  • internal/activator/claimwindow.go +50 −0
  • internal/activator/claimwindow_test.go +90 −0
  • internal/hostbox/restore.go +359 −0
  • internal/hostbox/restore_test.go +314 −0
  • internal/hostbox/wakepolicy.go +90 −0
  • internal/hostbox/wakepolicy_test.go +99 −0
  • internal/hostbox/writer.go +3 −0
  • internal/lease/claimgeneration.go +156 −0
  • internal/lease/claimgeneration_external_test.go +19 −0
  • internal/lease/claimgeneration_test.go +196 −0
  • internal/lease/lease.go +14 −8
  • internal/moves/install_contract.go +22 −0
  • internal/provision/moveinputs_contract.go +10 −0
  • internal/schedule/claimreplace_test.go +43 −0
  • internal/schedule/claimtimers.go +18 −0

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index a1afa63f4..399bd7698 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -69,6 +69,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/registry"
)
@@ -86,10 +87,12 @@ const SidecarHeader = "X-Carlos-Sidecar"
// proxy-only behavior byte-identical: instances run under systemd, the edge
// only proxies.
type Config struct {
- Bucket string // CARLOS_HIBERNATE_BUCKET (required to enable)
- Region string // CARLOS_HIBERNATE_REGION (optional; AWS default chain resolves "")
- Prefix string // CARLOS_HIBERNATE_PREFIX (optional key prefix inside the bucket)
- Idle time.Duration // CARLOS_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s")
+ HostWrites hostwrite.Config
+ Claims *ClaimRuntime
+ Bucket string // CARLOS_HIBERNATE_BUCKET (required to enable)
+ Region string // CARLOS_HIBERNATE_REGION (optional; AWS default chain resolves "")
+ Prefix string // CARLOS_HIBERNATE_PREFIX (optional key prefix inside the bucket)
+ Idle time.Duration // CARLOS_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s")
// AgentSocket is the unix socket a worker child dials to reach back —
// the direction that makes a sidecar possible at all: the agent listens,
// children call in; the platform never dials a sidecar. Handed to every
@@ -341,6 +344,7 @@ const (
)
type actEntry struct {
+ claimResult *wokenInstance
// bounceAfterWake is a Bounce that arrived while this INSTANCE was still
// waking. The wake had already read its config and execed its build, so
// the bounce's reason has not reached the child; wake() spends it the
@@ -1091,6 +1095,9 @@ type wokenInstance struct {
// fixed at entry construction, before the wake goroutine starts, so reading
// them here without a.mu is safe by construction.
func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, error) {
+ if a.cfg.HostWrites.Enabled {
+ return a.doClaimWake(ctx, e)
+ }
rt := e.route
host := rt.Host
w := &wokenInstance{stopOutcome: &stopObservation{}}
diff --git a/internal/activator/claimspawn.go b/internal/activator/claimspawn.go
index 75a5eb498..5282ace5e 100644
--- a/internal/activator/claimspawn.go
+++ b/internal/activator/claimspawn.go
@@ -9,6 +9,10 @@ import (
)
func spawnClaimed(ctx context.Context, enabled bool, manager *cgroup.Manager, host string, limit cgroup.Limit, selected hostwrite.Work, identity hostwrite.Identity, build func() *exec.Cmd) (*exec.Cmd, *cgroup.Prepared, error) {
+ return spawnClaimedBefore(ctx, enabled, manager, host, limit, selected, identity, build, nil)
+}
+
+func spawnClaimedBefore(ctx context.Context, enabled bool, manager *cgroup.Manager, host string, limit cgroup.Limit, selected hostwrite.Work, identity hostwrite.Identity, build func() *exec.Cmd, before func() error) (*exec.Cmd, *cgroup.Prepared, error) {
if !enabled {
cmd, err := manager.Start(host, limit, build)
return cmd, nil, err
@@ -24,6 +28,14 @@ func spawnClaimed(ctx context.Context, enabled bool, manager *cgroup.Manager, ho
if err != nil {
return err
}
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if before != nil {
+ if err := before(); err != nil {
+ return err
+ }
+ }
cmd = build()
return prepared.Start(cmd)
})
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
new file mode 100644
index 000000000..bc27f58d5
--- /dev/null
+++ b/internal/activator/claimwake.go
@@ -0,0 +1,334 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os/exec"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+var ErrAttemptOver = moves.ErrAttemptOver
+
+type StartPolicy struct {
+ MaxStarts int
+ Window time.Duration
+ Deadline time.Time
+ Probe bool
+ Prepared *moves.InstallManifest
+ Inputs *provision.MoveInputs
+}
+type startPolicy = StartPolicy
+
+// ClaimRuntime is assembled by the host owner before Service.Start.
+// Admit reserves the first serving attempt and binds its resulting cold session.
+// Prepared probes instead consume the moves owner's single-use admission.
+type ClaimRuntime struct {
+ FailedAdmission func(context.Context, hostclaim.Observed, error) error
+ Box string
+ Read func(context.Context, string) (hostclaim.Observed, error)
+ StopForeign func(context.Context, string) error
+ Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
+ Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
+ Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
+}
+
+// ClaimStart is one admitted session. Its closures retain the admission tuple.
+// Only Restore may purge data. Process retries reuse it and the live replication.
+type ClaimStart struct {
+ // OpenProbe journals the first app-start time before execution and returns
+ // that same time on retries, including after controller restart. Its
+ // provider binds the moves attempt and admission, never resets Opened.
+ OpenProbe func(context.Context, time.Time) (time.Time, error)
+ Work hostwrite.Work
+ Identity hostwrite.Identity
+ Manager *cgroup.Manager
+ Host string
+ Limit cgroup.Limit
+ Restore func(context.Context) error
+ Materialize func(context.Context) error
+ Build func() *exec.Cmd
+ Ready func(context.Context, *exec.Cmd) error
+ // Reserve(0) validates the already reserved first attempt. Later calls CAS.
+ Reserve func(context.Context, int) error
+ Failed func(context.Context, int) error
+ Healthy func(context.Context, int, *exec.Cmd) error
+ Replicate func(context.Context, string) (Rep, error)
+ Publish func(context.Context, hostclaim.Observed) (hostclaim.Observed, error)
+ // Retain owns uncertain process or replica cleanup until it can be proved.
+ Retain func(Proc, Rep, error)
+ Version string
+}
+
+func (s ClaimStart) valid() bool {
+ return s.Work != nil && s.Identity.Config != "" && s.Identity.Release != "" && s.Host != "" &&
+ s.Restore != nil && s.Materialize != nil && s.Build != nil && s.Ready != nil && s.Reserve != nil &&
+ s.Failed != nil && s.Healthy != nil && s.Replicate != nil && s.Publish != nil && s.Retain != nil
+}
+
+func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, error) {
+ r := a.cfg.Claims
+ if r == nil || r.Box == "" || r.Read == nil || r.StopForeign == nil {
+ return nil, hostwrite.ErrUnavailable
+ }
+ observed, err := r.Read(ctx, e.host)
+ if err != nil {
+ return nil, err
+ }
+ // K6 precedes registry reads, lease acquisition, materialization and policy checks.
+ if observed.Claim == nil || observed.Claim.Op.Exec != r.Box {
+ stopErr := r.StopForeign(ctx, e.host)
+ return nil, errors.Join(hostbox.ErrChanged, stopErr)
+ }
+ if err := hostbox.WakePhase(observed.Claim, r.Box, false); err != nil {
+ return nil, err
+ }
+ if r.Prerequisites == nil || r.Admit == nil || r.Prepare == nil || r.FailedAdmission == nil {
+ return nil, hostwrite.ErrUnavailable
+ }
+ rt := e.route
+ if a.Routes != nil {
+ fresh, err := a.Routes.Get(e.host)
+ if err != nil {
+ return nil, err
+ }
+ if fresh == nil {
+ return nil, hostbox.ErrChanged
+ }
+ rt = *fresh
+ }
+ policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second}
+ if err := r.Prerequisites(ctx, rt, observed, policy); err != nil {
+ return nil, err
+ }
+ admitted, err := r.Admit(ctx, rt, observed, policy)
+ if err != nil {
+ return nil, err
+ }
+ if admitted.Claim == nil || admitted.Claim.Holder == nil || observed.Claim.Holder == nil || admitted.Claim.Holder.Session <= observed.Claim.Holder.Session || admitted.Claim.Holder.Generation == observed.Claim.Holder.Generation {
+ return nil, hostbox.ErrChanged
+ }
+ e.route = rt
+ if err := a.startAdmitted(ctx, e, admitted, policy); err != nil {
+ recoveryCtx, cancelRecovery := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancelRecovery()
+ return nil, errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err))
+ }
+ return e.claimResult, nil
+}
+
+func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostclaim.Observed, policy startPolicy) (result error) {
+ r := a.cfg.Claims
+ if !a.cfg.HostWrites.Enabled || r == nil || r.Read == nil || r.Prepare == nil || e == nil || admitted.Host != e.host || policy.MaxStarts < 1 || policy.MaxStarts > 5 || policy.Window <= 0 || policy.Window > 30*time.Second {
+ return hostwrite.ErrUnavailable
+ }
+ budget := &probeWindow{policy: policy, now: time.Now}
+ defer func() { result = budget.result(result) }()
+ if !policy.Deadline.IsZero() {
+ bounded, cancel := context.WithDeadline(ctx, policy.Deadline)
+ defer cancel()
+ ctx = bounded
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if err := hostbox.WakePhase(admitted.Claim, r.Box, policy.Probe); err != nil {
+ return err
+ }
+ fresh, err := r.Read(ctx, e.host)
+ if err != nil {
+ return err
+ }
+ if fresh.ETag != admitted.ETag {
+ return hostbox.ErrChanged
+ }
+ s, err := r.Prepare(ctx, e.route, admitted, policy)
+ if err != nil {
+ return err
+ }
+ if !s.valid() || s.Host != e.host || (policy.Probe && s.OpenProbe == nil) {
+ return hostwrite.ErrUnavailable
+ }
+ prefix, err := hostbox.GenerationPrefix(*admitted.Claim.Holder)
+ if err != nil {
+ return err
+ }
+ if err := s.Restore(ctx); err != nil {
+ return err
+ }
+ var rep Rep
+ var retained bool
+ defer func() {
+ if result == nil || rep == nil || retained {
+ return
+ }
+ cleanup, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := rep.Close(cleanup); err != nil {
+ s.Retain(nil, rep, err)
+ result = errors.Join(result, err)
+ }
+ }()
+ var last error
+ for attempt := 0; attempt < policy.MaxStarts; attempt++ {
+ if err := budget.result(ctx.Err()); err != nil {
+ return err
+ }
+ if err := s.Reserve(ctx, attempt); err != nil {
+ return err
+ }
+ if err := s.Materialize(ctx); err != nil {
+ return errors.Join(err, s.Failed(ctx, attempt))
+ }
+ // This hook runs after strict preparation and immediately before Start.
+ // Restore and materialization consume only the separate cut deadline.
+ beforeStart := func() error { return budget.start(ctx, s.OpenProbe) }
+ cmd, prepared, startErr := spawnClaimedBefore(ctx, true, s.Manager, s.Host, s.Limit, s.Work, s.Identity, s.Build, beforeStart)
+
+ p := newClaimProcess(cmd, prepared)
+ if startErr == nil {
+ deadline := time.Now().Add(policy.Window)
+ if policy.Probe {
+ deadline = budget.opened.Add(policy.Window)
+ }
+ readyCtx, cancelReady := context.WithDeadline(ctx, deadline)
+ startErr = s.Ready(readyCtx, cmd)
+ cancelReady()
+ startErr = budget.result(startErr)
+ }
+ ready := startErr == nil
+ budget.ready = ready
+ if ready {
+ rep, startErr = s.Replicate(ctx, prefix)
+ if startErr == nil && rep == nil {
+ startErr = hostwrite.ErrUnavailable
+ }
+ }
+ if startErr == nil {
+ current, err := r.Read(ctx, e.host)
+ if err != nil {
+ startErr = err
+ } else if !sameAdmission(current, admitted) {
+ startErr = hostbox.ErrChanged
+ } else {
+ _, startErr = s.Publish(ctx, current)
+ }
+ }
+ if startErr == nil {
+ startErr = s.Healthy(ctx, attempt, cmd)
+ }
+ if startErr == nil {
+ e.claimResult = &wokenInstance{proc: p, rep: rep, spawnVersion: s.Version, stopOutcome: &stopObservation{}}
+ return nil
+ }
+ cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second)
+ cleanErr := p.Stop(cleanup)
+ cancelCleanup()
+ if cleanErr != nil {
+ s.Retain(p, rep, errors.Join(startErr, cleanErr))
+ retained = true
+ return errors.Join(startErr, cleanErr)
+ }
+ if err := s.Failed(ctx, attempt); err != nil {
+ return errors.Join(startErr, err)
+ }
+ last = startErr
+ // A replication/publication failure has an uncertain remote outcome.
+ // The next process start cannot reuse that opening as another attempt.
+ if ready {
+ return last
+ }
+ if errors.Is(startErr, ErrAttemptOver) {
+ return startErr
+ }
+ if attempt+1 < policy.MaxStarts {
+ waitCtx := ctx
+ cancelWait := func() {}
+ if policy.Probe && !budget.opened.IsZero() {
+ waitCtx, cancelWait = context.WithDeadline(ctx, budget.opened.Add(policy.Window))
+ }
+ timer := time.NewTimer(time.Second << attempt)
+ select {
+ case <-waitCtx.Done():
+ timer.Stop()
+ cancelWait()
+ return budget.result(waitCtx.Err())
+ case <-timer.C:
+ cancelWait()
+ }
+ }
+ }
+ return last
+}
+
+type claimProcess struct {
+ cmd *exec.Cmd
+ prepared *cgroup.Prepared
+ done chan struct{}
+ mu sync.Mutex
+ drained, closed bool
+}
+
+func newClaimProcess(cmd *exec.Cmd, p *cgroup.Prepared) *claimProcess {
+ c := &claimProcess{cmd: cmd, prepared: p, done: make(chan struct{})}
+ if cmd == nil || cmd.Process == nil {
+ close(c.done)
+ } else {
+ go func() { _ = cmd.Wait(); close(c.done) }()
+ }
+ return c
+}
+func (p *claimProcess) Exited() bool {
+ select {
+ case <-p.done:
+ return true
+ default:
+ return false
+ }
+}
+func (p *claimProcess) Stop(ctx context.Context) error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.closed {
+ return nil
+ }
+ if p.prepared == nil {
+ if p.cmd != nil && p.cmd.Process != nil {
+ return hostwrite.ErrUnavailable
+ }
+ p.closed = true
+ return nil
+ }
+ if !p.drained {
+ if err := p.prepared.KillAndEmpty(ctx); err != nil {
+ return err
+ }
+ p.drained = true
+ }
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-p.done:
+ }
+ if err := p.prepared.Close(); err != nil {
+ return err
+ }
+ p.closed = true
+ return nil
+}
+
+func sameAdmission(a, b hostclaim.Observed) bool {
+ if a.Absent || b.Absent || a.Host != b.Host || a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil {
+ return false
+ }
+ x, y := a.Claim, b.Claim
+ return x.Holder.Incarnation == y.Holder.Incarnation && x.Holder.Stamp == y.Holder.Stamp && x.Holder.Session == y.Holder.Session && x.Holder.Generation == y.Holder.Generation && x.Op.ID == y.Op.ID && x.Op.Kind == y.Op.Kind && x.Op.Phase == y.Op.Phase && x.Op.Exec == y.Op.Exec
+}
diff --git a/internal/activator/claimwake_test.go b/internal/activator/claimwake_test.go
new file mode 100644
index 000000000..9b176678c
--- /dev/null
+++ b/internal/activator/claimwake_test.go
@@ -0,0 +1,75 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+ "testing"
+ "time"
+)
+
+type claimRoutes struct{ called *bool }
+
+func (r claimRoutes) Get(string) (*registry.Route, error) {
+ *r.called = true
+ return nil, errors.New("unexpected registry read")
+}
+func TestClaimK6StopsBeforeRegistryOrAdmission(t *testing.T) {
+ stopped, readRoute, admitted := false, false, false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{
+ Box: "here", Read: func(context.Context, string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{Claim: &hostclaim.Claim{Op: hostclaim.Op{Exec: "elsewhere"}}}, nil
+ },
+ StopForeign: func(context.Context, string) error { stopped = true; return nil },
+ Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
+ admitted = true
+ return hostclaim.Observed{}, nil
+ },
+ }}, Routes: claimRoutes{&readRoute}}
+ if _, err := a.doWake(context.Background(), &actEntry{host: "x.example"}); err == nil {
+ t.Fatal("foreign wake")
+ }
+ if !stopped || readRoute || admitted {
+ t.Fatalf("stop %v route %v admit %v", stopped, readRoute, admitted)
+ }
+}
+func TestClaimReadFailureDoesNotInventForeignAuthority(t *testing.T) {
+ stopped := false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Read: func(context.Context, string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{}, errors.New("offline")
+ }, StopForeign: func(context.Context, string) error { stopped = true; return nil }}}}
+ if _, err := a.doWake(context.Background(), &actEntry{host: "x.example"}); err == nil || stopped {
+ t.Fatal(err, stopped)
+ }
+}
+func TestStartAdmittedRejectsMovedObservationBeforeRestore(t *testing.T) {
+ prepared := false
+ obs := hostclaim.Observed{Host: "x.example", ETag: "old", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 1, Generation: "g"}, Op: hostclaim.Op{Kind: "landing", ID: "op", Phase: "probing", Exec: "box"}}}
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Read: func(context.Context, string) (hostclaim.Observed, error) { n := obs; n.ETag = "new"; return n, nil }, Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ prepared = true
+ return ClaimStart{}, nil
+ }}}}
+ err := a.startAdmitted(context.Background(), &actEntry{host: obs.Host}, obs, startPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true})
+ if err == nil || prepared {
+ t.Fatal(err, prepared)
+ }
+}
+func TestSameAdmissionAllowsBookkeepingButRefusesSessionChange(t *testing.T) {
+ h := &hostclaim.Holder{Incarnation: "i", Stamp: 2, Session: 2, Generation: "g"}
+ a := hostclaim.Observed{Host: "x.example", ETag: "a", Claim: &hostclaim.Claim{Holder: h, Op: hostclaim.Op{Kind: "serving", Exec: "box"}}}
+ b := a
+ b.ETag = "b"
+ if !sameAdmission(a, b) {
+ t.Fatal("bookkeeping")
+ }
+ c := *a.Claim
+ holder := *h
+ holder.Session++
+ c.Holder = &holder
+ b.Claim = &c
+ if sameAdmission(a, b) {
+ t.Fatal("session")
+ }
+}
diff --git a/internal/activator/claimwindow.go b/internal/activator/claimwindow.go
new file mode 100644
index 000000000..08abe7739
--- /dev/null
+++ b/internal/activator/claimwindow.go
@@ -0,0 +1,50 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+type probeWindow struct {
+ policy startPolicy
+ opened time.Time
+ ready bool
+ now func() time.Time
+}
+
+func (b *probeWindow) result(err error) error {
+ if !b.policy.Probe || b.ready {
+ return err
+ }
+ now := b.now()
+ if (!b.policy.Deadline.IsZero() && !now.Before(b.policy.Deadline)) ||
+ (!b.opened.IsZero() && !now.Before(b.opened.Add(b.policy.Window))) {
+ return errors.Join(ErrAttemptOver, err)
+ }
+ return err
+}
+
+func (b *probeWindow) start(ctx context.Context, journal func(context.Context, time.Time) (time.Time, error)) error {
+ if !b.policy.Probe {
+ return ctx.Err()
+ }
+ if err := b.result(ctx.Err()); err != nil {
+ return err
+ }
+ if journal == nil {
+ return hostwrite.ErrUnavailable
+ }
+ proposed := b.now()
+ opened, err := journal(ctx, proposed)
+ if err != nil {
+ return err
+ }
+ if opened.IsZero() || opened.After(proposed) || (!b.opened.IsZero() && !b.opened.Equal(opened)) {
+ return hostwrite.ErrUnavailable
+ }
+ b.opened = opened
+ return b.result(ctx.Err())
+}
diff --git a/internal/activator/claimwindow_test.go b/internal/activator/claimwindow_test.go
new file mode 100644
index 000000000..82338e25a
--- /dev/null
+++ b/internal/activator/claimwindow_test.go
@@ -0,0 +1,90 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "os/exec"
+ "testing"
+ "time"
+)
+
+func TestStartAdmittedPreparedInputsAndRestoreOutsideProbeWindow(t *testing.T) {
+ manifest := &moves.InstallManifest{Op: "op", Host: "x.example", SnapshotVersion: "pinned"}
+ inputs := &provision.MoveInputs{Release: "release", OverlayETag: "overlay"}
+ policy := startPolicy{MaxStarts: 5, Window: time.Millisecond, Deadline: time.Now().Add(time.Minute), Probe: true, Prepared: manifest, Inputs: inputs}
+ obs := hostclaim.Observed{Host: "x.example", ETag: "one", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Location: "instances/x.example/inc", Session: 1, Generation: "one"}, Op: hostclaim.Op{Kind: "landing", ID: "op", Phase: "probing", Exec: "box"}}}
+ work, _ := (hostwrite.Config{}).Select(t.Context(), obs.Host, "spawn")
+ stop := errors.New("stop before kernel start")
+ restored, materialized, journaled := false, false, false
+ a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil }, Prepare: func(ctx context.Context, rt registry.Route, got hostclaim.Observed, p startPolicy) (ClaimStart, error) {
+ if p.Prepared != manifest || p.Inputs != inputs {
+ t.Fatal("C1 inputs lost before preparation")
+ }
+ return ClaimStart{Host: obs.Host, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"},
+ OpenProbe: func(context.Context, time.Time) (time.Time, error) { journaled = true; return time.Now(), nil },
+ Restore: func(ctx context.Context) error {
+ restored = true
+ deadline, ok := ctx.Deadline()
+ if !ok || !deadline.Equal(policy.Deadline) {
+ t.Fatal("restore consumed probe window")
+ }
+ time.Sleep(5 * policy.Window)
+ return ctx.Err()
+ },
+ Materialize: func(context.Context) error { materialized = true; return stop }, Build: func() *exec.Cmd { return nil }, Ready: func(context.Context, *exec.Cmd) error { return nil },
+ Reserve: func(context.Context, int) error { return nil }, Failed: func(context.Context, int) error { return nil }, Healthy: func(context.Context, int, *exec.Cmd) error { return nil }, Replicate: func(context.Context, string) (Rep, error) { return nil, nil }, Publish: func(context.Context, hostclaim.Observed) (hostclaim.Observed, error) { return obs, nil }, Retain: func(Proc, Rep, error) {},
+ }, nil
+ }}}}
+ err := a.startAdmitted(t.Context(), &actEntry{host: obs.Host}, obs, policy)
+ if !errors.Is(err, stop) || !restored || !materialized || journaled {
+ t.Fatalf("err=%v restore=%v materialize=%v journal=%v", err, restored, materialized, journaled)
+ }
+ policy.Deadline = time.Now().Add(-time.Second)
+ if err := a.startAdmitted(t.Context(), &actEntry{host: obs.Host}, obs, policy); !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("cut deadline: %v", err)
+ }
+}
+
+func TestClaimProbeWindowJournalsFirstStartAndSurvivesRetry(t *testing.T) {
+ now := time.Unix(1000, 0)
+ var saved time.Time
+ policy := startPolicy{Probe: true, Window: 30 * time.Second, Deadline: now.Add(time.Minute)}
+ journal := func(_ context.Context, proposed time.Time) (time.Time, error) {
+ if saved.IsZero() {
+ saved = proposed
+ }
+ return saved, nil
+ }
+ b := &probeWindow{policy: policy, now: func() time.Time { return now }}
+ // Restore took longer than the window, but has not started the app.
+ now = now.Add(40 * time.Second)
+ if err := b.start(t.Context(), journal); err != nil {
+ t.Fatal(err)
+ }
+ first := saved
+ now = now.Add(time.Second)
+ if err := b.start(t.Context(), journal); err != nil || saved != first {
+ t.Fatal("retry reset", err, saved)
+ }
+ b = &probeWindow{policy: policy, now: func() time.Time { return now }}
+ if err := b.start(t.Context(), journal); err != nil || b.opened != first {
+ t.Fatal("restart reset", err, b.opened)
+ }
+ now = policy.Deadline
+ if !errors.Is(b.result(nil), ErrAttemptOver) {
+ t.Fatal("late readiness accepted after cut")
+ }
+ b.policy.Deadline = time.Time{}
+ now = first.Add(policy.Window)
+ if err := b.result(nil); !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("late readiness ignored context: %v", err)
+ }
+ if err := b.start(t.Context(), journal); !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("expired retry: %v", err)
+ }
+}
diff --git a/internal/hostbox/restore.go b/internal/hostbox/restore.go
new file mode 100644
index 000000000..c4a606bd4
--- /dev/null
+++ b/internal/hostbox/restore.go
@@ -0,0 +1,359 @@
+package hostbox
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+
+ "github.com/benbjohnson/litestream"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/superfly/ltx"
+)
+
+// RetireArtifacts moves a predecessor installation out of the live namespace.
+// It is called only when the durable local generation differs from the selected
+// generation, and while Work still vouches for the installation being replaced.
+type RetireArtifacts func(context.Context, string) error
+
+// ColdRestore installs a selected immutable generation. It never accepts an
+// unselected current object and never reuses a local database.
+type ColdRestore struct {
+ Reader genselect.Reader
+ Bucket string
+ Generation string // newly admitted installation, distinct from baseline plan.Generation
+ Work hostwrite.Work
+ Identity hostwrite.Identity
+ RetireArtifacts RetireArtifacts
+}
+
+// Install purges the old local state and restores plan into dbPath. All local
+// mutation stays inside the work selected after admission.
+func (r ColdRestore) Install(ctx context.Context, dbPath string, plan *genselect.Plan) error {
+ if ctx == nil || r.Reader == nil || r.Bucket == "" || r.Work == nil || plan == nil ||
+ plan.Generation == "" || !safeGeneration(r.Generation) || len(plan.Files) == 0 || plan.MaxTXID == 0 || dbPath == "" {
+ return ErrUnsupported
+ }
+ var capturedGeneration string
+ if err := r.Work.With(ctx, r.Identity, func() error {
+ var err error
+ capturedGeneration, err = readGeneration(dbPath + ".gen")
+ return err
+ }); err != nil {
+ return fmt.Errorf("capture local generation: %w", err)
+ }
+ files, finalChecksum, err := loadPinned(ctx, r.Reader, r.Bucket, plan)
+ if err != nil {
+ return err
+ }
+ // Retirement takes park.Local's own flock. Calling it inside Work.With
+ // would recursively acquire the same lock and deadlock.
+ if capturedGeneration != "" && capturedGeneration != r.Generation {
+ if r.RetireArtifacts == nil {
+ return fmt.Errorf("retire generation %q: %w", capturedGeneration, ErrUnsupported)
+ }
+ if err := r.RetireArtifacts(ctx, capturedGeneration); err != nil {
+ return fmt.Errorf("retire generation %q: %w", capturedGeneration, err)
+ }
+ }
+ return r.Work.With(ctx, r.Identity, func() error {
+ currentGeneration, err := readGeneration(dbPath + ".gen")
+ if err != nil {
+ return fmt.Errorf("recheck local generation: %w", err)
+ }
+ if capturedGeneration != "" && capturedGeneration != r.Generation {
+ // Successful predecessor retirement normally moves the enrolled
+ // marker with the rest of the installation. Absence is therefore the
+ // expected receipt disposition. A different marker means a successor
+ // appeared after our capture and must never be purged.
+ if currentGeneration != "" {
+ return ErrChanged
+ }
+ } else if currentGeneration != capturedGeneration {
+ return ErrChanged
+ }
+ if err := purgeDB(dbPath); err != nil {
+ return err
+ }
+ tmp, err := os.CreateTemp(filepath.Dir(dbPath), ".restore-*.db")
+ if err != nil {
+ return fmt.Errorf("create restore temp: %w", err)
+ }
+ tmpPath := tmp.Name()
+ if err := tmp.Close(); err != nil {
+ _ = os.Remove(tmpPath)
+ return err
+ }
+ _ = os.Remove(tmpPath) // Litestream requires an absent output.
+ defer os.Remove(tmpPath)
+ client := &pinnedClient{files: files}
+ replica := litestream.NewReplicaWithClient(nil, client)
+ opt := litestream.NewRestoreOptions()
+ opt.OutputPath = tmpPath
+ opt.TXID = ltx.TXID(plan.MaxTXID)
+ if err := replica.Restore(ctx, opt); err != nil {
+ return fmt.Errorf("restore selected generation: %w", err)
+ }
+ f, err := os.Open(tmpPath)
+ if err != nil {
+ return err
+ }
+ checksum, checkErr := ltx.ChecksumReader(f, files[0].infoPageSize)
+ closeErr := f.Close()
+ if checkErr != nil {
+ return fmt.Errorf("checksum restored database: %w", checkErr)
+ }
+ if closeErr != nil {
+ return closeErr
+ }
+ // HeaderFlagNoChecksum is Litestream's pinned on-disk format. Its LTX
+ // trailer still detects corruption across the complete file, but is not
+ // an authenticity proof and intentionally has no database checksum to
+ // compare. A moves-owned object digest must supply authenticity. When
+ // the selected chain carries a database checksum, require the restored
+ // database to match it.
+ if finalChecksum != 0 && checksum != finalChecksum {
+ return fmt.Errorf("restored database checksum %s, want %s", checksum, finalChecksum)
+ }
+ if err := lease.CheckDBContext(ctx, tmpPath); err != nil {
+ return fmt.Errorf("check restored database: %w", err)
+ }
+ if err := syncFile(tmpPath); err != nil {
+ return err
+ }
+ if err := os.Rename(tmpPath, dbPath); err != nil {
+ return fmt.Errorf("install restored database: %w", err)
+ }
+ if err := writeGeneration(dbPath+".gen", r.Generation); err != nil {
+ return fmt.Errorf("install generation marker: %w", err)
+ }
+ return syncDir(filepath.Dir(dbPath))
+ })
+}
+
+// InstallEmpty installs a separately proved initial-empty admission. It writes
+// no database and no zero-TXID LTX. Authority must prove both the complete
+// empty remote baseline and first-creator status before local mutation.
+func (r ColdRestore) InstallEmpty(ctx context.Context, dbPath string, authority func(context.Context) error) error {
+ if ctx == nil || r.Work == nil || !safeGeneration(r.Generation) || dbPath == "" || authority == nil {
+ return ErrUnsupported
+ }
+ if err := authority(ctx); err != nil {
+ return err
+ }
+ var captured string
+ if err := r.Work.With(ctx, r.Identity, func() error { var e error; captured, e = readGeneration(dbPath + ".gen"); return e }); err != nil {
+ return err
+ }
+ if captured != "" && captured != r.Generation {
+ if r.RetireArtifacts == nil {
+ return ErrUnsupported
+ }
+ if err := r.RetireArtifacts(ctx, captured); err != nil {
+ return err
+ }
+ }
+ return r.Work.With(ctx, r.Identity, func() error {
+ current, err := readGeneration(dbPath + ".gen")
+ if err != nil {
+ return err
+ }
+ if captured != "" && captured != r.Generation {
+ if current != "" {
+ return ErrChanged
+ }
+ } else if current != captured {
+ return ErrChanged
+ }
+ if err := purgeDB(dbPath); err != nil {
+ return err
+ }
+ if err := writeGeneration(dbPath+".gen", r.Generation); err != nil {
+ return err
+ }
+ return syncDir(filepath.Dir(dbPath))
+ })
+}
+
+func safeGeneration(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00\r\n")
+}
+
+func purgeDB(path string) error {
+ for _, p := range []string{path, path + "-wal", path + "-shm", path + ".gen"} {
+ if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return fmt.Errorf("purge %s: %w", p, err)
+ }
+ }
+ for _, p := range []string{
+ path + "-litestream",
+ filepath.Join(filepath.Dir(path), ".activator-"+filepath.Base(path)+"-litestream"),
+ } {
+ if err := os.RemoveAll(p); err != nil {
+ return fmt.Errorf("purge %s: %w", p, err)
+ }
+ }
+ return nil
+}
+
+func readGeneration(path string) (string, error) {
+ b, err := os.ReadFile(path)
+ if errors.Is(err, os.ErrNotExist) {
+ return "", nil
+ }
+ if err != nil {
+ return "", err
+ }
+ g := strings.TrimSpace(string(b))
+ if g == "" || strings.ContainsAny(g, "/\\\x00\r\n") {
+ return "", errors.New("invalid generation marker")
+ }
+ return g, nil
+}
+
+func writeGeneration(path, generation string) error {
+ tmp, err := os.CreateTemp(filepath.Dir(path), ".generation-*")
+ if err != nil {
+ return err
+ }
+ name := tmp.Name()
+ defer os.Remove(name)
+ if _, err = io.WriteString(tmp, generation+"\n"); err == nil {
+ err = tmp.Sync()
+ }
+ if closeErr := tmp.Close(); err == nil {
+ err = closeErr
+ }
+ if err != nil {
+ return err
+ }
+ return os.Rename(name, path)
+}
+
+func syncFile(path string) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return f.Sync()
+}
+func syncDir(path string) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return f.Sync()
+}
+
+type pinnedFile struct {
+ plan genselect.File
+ data []byte
+ info *ltx.FileInfo
+ infoPageSize int
+}
+
+func loadPinned(ctx context.Context, reader genselect.Reader, bucket string, plan *genselect.Plan) ([]pinnedFile, ltx.Checksum, error) {
+ files := make([]pinnedFile, 0, len(plan.Files))
+ var last ltx.Checksum
+ var previous uint64
+ for i, selected := range plan.Files {
+ if i != 0 && (previous == ^uint64(0) || !ltx.IsContiguous(ltx.TXID(previous), ltx.TXID(selected.MinTXID), ltx.TXID(selected.MaxTXID))) {
+ return nil, 0, genselect.ErrGap
+ }
+ rc, err := reader.OpenVersion(ctx, bucket, selected.Key, selected.VersionID)
+ if err != nil {
+ return nil, 0, fmt.Errorf("open selected %s@%s: %w", selected.Key, selected.VersionID, err)
+ }
+ data, readErr := io.ReadAll(rc)
+ closeErr := rc.Close()
+ if readErr != nil {
+ return nil, 0, fmt.Errorf("read selected %s: %w", selected.Key, readErr)
+ }
+ if closeErr != nil {
+ return nil, 0, fmt.Errorf("close selected %s: %w", selected.Key, closeErr)
+ }
+ if int64(len(data)) != selected.Size {
+ return nil, 0, fmt.Errorf("selected %s size %d, want %d", selected.Key, len(data), selected.Size)
+ }
+ dec := ltx.NewDecoder(bytes.NewReader(data))
+ if err := dec.DecodeHeader(); err != nil {
+ return nil, 0, fmt.Errorf("decode %s header: %w", selected.Key, err)
+ }
+ h := dec.Header()
+ if uint64(h.MinTXID) != selected.MinTXID || uint64(h.MaxTXID) != selected.MaxTXID {
+ return nil, 0, fmt.Errorf("selected %s range mismatch", selected.Key)
+ }
+ buf := make([]byte, h.PageSize)
+ var ph ltx.PageHeader
+ for {
+ err = dec.DecodePage(&ph, buf)
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return nil, 0, fmt.Errorf("decode %s: %w", selected.Key, err)
+ }
+ }
+ if err := dec.Close(); err != nil {
+ return nil, 0, fmt.Errorf("verify %s trailer: %w", selected.Key, err)
+ }
+ t := dec.Trailer()
+ if i == 0 && plan.Pin != nil && plan.Pin.SnapshotCRC64 != 0 && uint64(t.FileChecksum) != plan.Pin.SnapshotCRC64 {
+ return nil, 0, genselect.ErrPinnedMismatch
+ }
+ info := &ltx.FileInfo{Level: selected.Level, MinTXID: h.MinTXID, MaxTXID: h.MaxTXID, PreApplyChecksum: h.PreApplyChecksum, PostApplyChecksum: t.PostApplyChecksum, Size: int64(len(data))}
+ files = append(files, pinnedFile{plan: selected, data: data, info: info, infoPageSize: int(h.PageSize)})
+ last, previous = t.PostApplyChecksum, selected.MaxTXID
+ }
+ if previous != plan.MaxTXID {
+ return nil, 0, fmt.Errorf("selected final TXID %d, want %d", previous, plan.MaxTXID)
+ }
+ return files, last, nil
+}
+
+type pinnedClient struct{ files []pinnedFile }
+
+func (*pinnedClient) Type() string { return "selected-versions" }
+func (*pinnedClient) Init(context.Context) error { return nil }
+func (*pinnedClient) SetLogger(*slog.Logger) {}
+func (c *pinnedClient) LTXFiles(_ context.Context, level int, seek ltx.TXID, _ bool) (ltx.FileIterator, error) {
+ var a []*ltx.FileInfo
+ for _, f := range c.files {
+ if f.info.Level == level && f.info.MaxTXID >= seek {
+ cp := *f.info
+ a = append(a, &cp)
+ }
+ }
+ sort.Slice(a, func(i, j int) bool { return a[i].MinTXID < a[j].MinTXID })
+ return ltx.NewFileInfoSliceIterator(a), nil
+}
+func (c *pinnedClient) OpenLTXFile(_ context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ for _, f := range c.files {
+ if f.info.Level == level && f.info.MinTXID == min && f.info.MaxTXID == max {
+ if offset < 0 || size < 0 || offset > int64(len(f.data)) {
+ return nil, os.ErrInvalid
+ }
+ end := int64(len(f.data))
+ if size > 0 && offset+size < end {
+ end = offset + size
+ }
+ return io.NopCloser(bytes.NewReader(f.data[offset:end])), nil
+ }
+ }
+ return nil, os.ErrNotExist
+}
+func (*pinnedClient) WriteLTXFile(context.Context, int, ltx.TXID, ltx.TXID, io.Reader) (*ltx.FileInfo, error) {
+ return nil, ErrUnsupported
+}
+func (*pinnedClient) DeleteLTXFiles(context.Context, []*ltx.FileInfo) error { return ErrUnsupported }
+func (*pinnedClient) DeleteAll(context.Context) error { return ErrUnsupported }
diff --git a/internal/hostbox/restore_test.go b/internal/hostbox/restore_test.go
new file mode 100644
index 000000000..0648fcd86
--- /dev/null
+++ b/internal/hostbox/restore_test.go
@@ -0,0 +1,314 @@
+package hostbox
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "encoding/binary"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/superfly/ltx"
+ _ "modernc.org/sqlite"
+)
+
+type restoreReader struct {
+ data map[string][]byte
+ openedKey string
+ openedVers string
+}
+
+func (*restoreReader) ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error) {
+ return nil, errors.New("unexpected listing")
+}
+func (*restoreReader) GetVersion(context.Context, string, string, string) ([]byte, error) {
+ return nil, errors.New("unexpected get")
+}
+func (r *restoreReader) OpenVersion(_ context.Context, _ string, key, version string) (io.ReadCloser, error) {
+ r.openedKey, r.openedVers = key, version
+ b, ok := r.data[key+"@"+version]
+ if !ok {
+ return nil, errors.New("missing immutable version")
+ }
+ return io.NopCloser(bytes.NewReader(b)), nil
+}
+
+func TestLoadPinnedReadsTheSelectedVersionAndRejectsCorruption(t *testing.T) {
+ r := &restoreReader{data: map[string][]byte{
+ "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx@selected": []byte("corrupt selected object"),
+ "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx@current": []byte("different current object"),
+ }}
+ p := &genselect.Plan{Generation: "one", MaxTXID: 1, Files: []genselect.File{{
+ Key: "instances/shop/g-one/0009/0000000000000001-0000000000000001.ltx", VersionID: "selected", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len("corrupt selected object")),
+ }}}
+ if _, _, err := loadPinned(context.Background(), r, "bucket", p); err == nil {
+ t.Fatal("loadPinned accepted an object without a valid complete LTX trailer")
+ }
+ if r.openedVers != "selected" {
+ t.Fatalf("opened version %q, want immutable selected version", r.openedVers)
+ }
+}
+
+type restoreWork struct {
+ calls int
+ active bool
+}
+
+func (w *restoreWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ w.calls++
+ w.active = true
+ defer func() { w.active = false }()
+ return fn()
+}
+
+func TestColdRestoreRefusesBeforeMutationWithoutSelectedAuthority(t *testing.T) {
+ r := ColdRestore{}
+ if err := r.Install(context.Background(), t.TempDir()+"/db", &genselect.Plan{}); !errors.Is(err, ErrUnsupported) {
+ t.Fatalf("Install error = %v, want ErrUnsupported", err)
+ }
+}
+
+func TestColdRestorePurgesSameGenerationUnderSelectedWork(t *testing.T) {
+ dir := t.TempDir()
+ db := filepath.Join(dir, "shop.db")
+ if err := os.WriteFile(db, []byte("warm database must not be reused"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(db+".gen", []byte("one\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ key, snapshot, crc := snapshotFixture(t, "selected")
+ rd := &restoreReader{data: map[string][]byte{key + "@v1": snapshot}}
+ work := &restoreWork{}
+ err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "one", Work: work}).Install(context.Background(), db,
+ &genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: crc}, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(snapshot))}}})
+ if err != nil {
+ t.Fatalf("Install: %v", err)
+ }
+ if work.calls != 2 {
+ t.Fatalf("selected work calls = %d, want capture and install", work.calls)
+ }
+ dbcheck, err := sql.Open("sqlite", db)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer dbcheck.Close()
+ var got string
+ if err := dbcheck.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "selected" {
+ t.Fatalf("restored value=%q err=%v", got, err)
+ }
+ if string(mustRead(t, db+".gen")) != "one\n" {
+ t.Fatal("generation marker not installed")
+ }
+}
+
+func TestColdRestorePinsVersionAndRetiresForeignGenerationOutsideWork(t *testing.T) {
+ dir := t.TempDir()
+ db := filepath.Join(dir, "shop.db")
+ if err := os.WriteFile(db, []byte("foreign"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(db+".gen", []byte("old\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ key, v1, crc := snapshotFixture(t, "selected-v1")
+ _, v2, _ := snapshotFixture(t, "current-v2")
+ rd := &restoreReader{data: map[string][]byte{key + "@v1": v1, key + "@v2": v2}}
+ work := &restoreWork{}
+ retired := ""
+ parked := filepath.Join(dir, "parked")
+ err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "new", Work: work, RetireArtifacts: func(_ context.Context, g string) error {
+ if work.active {
+ t.Fatal("retirement ran under Work flock")
+ }
+ retired = g
+ if err := os.Mkdir(parked, 0o700); err != nil {
+ return err
+ }
+ if err := os.Rename(db, filepath.Join(parked, "shop.db")); err != nil {
+ return err
+ }
+ return os.Rename(db+".gen", filepath.Join(parked, "shop.db.gen"))
+ }}).Install(context.Background(), db,
+ &genselect.Plan{Generation: "one", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: crc}, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(v1))}}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if retired != "old" {
+ t.Fatalf("retired %q, want old", retired)
+ }
+ if rd.openedVers != "v1" {
+ t.Fatalf("opened %q, want pinned v1", rd.openedVers)
+ }
+ check, err := sql.Open("sqlite", db)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer check.Close()
+ var got string
+ if err := check.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "selected-v1" {
+ t.Fatalf("restored value=%q err=%v", got, err)
+ }
+ if string(mustRead(t, db+".gen")) != "new\n" {
+ t.Fatalf("installed marker=%q, want admitted generation", mustRead(t, db+".gen"))
+ }
+}
+
+func TestColdRestoreReportsPurgeFailureBeforeInstall(t *testing.T) {
+ dir := t.TempDir()
+ db := filepath.Join(dir, "shop.db")
+ if err := os.Mkdir(db, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(db, "still-here"), []byte("x"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ key, snapshot, _ := snapshotFixture(t, "never-installed")
+ rd := &restoreReader{data: map[string][]byte{key + "@v1": snapshot}}
+ err := (ColdRestore{Reader: rd, Bucket: "bucket", Generation: "one", Work: &restoreWork{}}).Install(context.Background(), db,
+ &genselect.Plan{Generation: "one", MaxTXID: 1, Files: []genselect.File{{Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(len(snapshot))}}})
+ if err == nil || !strings.Contains(err.Error(), "purge "+db) {
+ t.Fatalf("Install error=%v, want checked purge failure", err)
+ }
+ if _, err := os.Stat(filepath.Join(db, "still-here")); err != nil {
+ t.Fatalf("purge failure damaged existing artifact: %v", err)
+ }
+}
+
+func TestColdRestoreInstallEmptyRequiresProofAndWritesOnlyGeneration(t *testing.T) {
+ dir := t.TempDir()
+ db := filepath.Join(dir, "empty.db")
+ work := &restoreWork{}
+ proved := false
+ r := ColdRestore{Generation: "new", Work: work}
+ if err := r.InstallEmpty(context.Background(), db, func(context.Context) error { proved = true; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ if !proved || work.calls != 2 {
+ t.Fatalf("proved=%v guarded calls=%d", proved, work.calls)
+ }
+ if _, err := os.Stat(db); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("empty bootstrap wrote DB: %v", err)
+ }
+ if string(mustRead(t, db+".gen")) != "new\n" {
+ t.Fatal("admitted generation missing")
+ }
+}
+
+func TestColdRestoreAcceptsSelectorOverlappingOutput(t *testing.T) {
+ for _, min := range []ltx.TXID{10, 12} {
+ t.Run(fmt.Sprintf("continuation-%d-13", min), func(t *testing.T) {
+ ctx := t.Context()
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ bucket.PageSize = 1
+ key, snapshot, crc := restoreLTXFixture(t, "snapshot-at-11", 9, 1, 11)
+ version := bucket.Put("data", key, snapshot)
+ continuation, data, _ := restoreLTXFixture(t, "compacted-at-13", 1, min, 13)
+ bucket.Put("data", continuation, data)
+ plan, err := genselect.SelectPrepared(ctx, bucket, genselect.Request{
+ Bucket: "data", Region: "ie", Host: "shop",
+ Holder: hostclaim.Holder{Incarnation: "inc", Location: "instances/shop"},
+ }, genselect.PreparedPin{Generation: "one", SnapshotKey: key, SnapshotVersion: version, SnapshotTXID: 11, SnapshotCRC64: crc})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(plan.Files) != 2 || plan.Files[0].MinTXID != 1 || plan.Files[0].MaxTXID != 11 || plan.Files[1].MinTXID != uint64(min) || plan.Files[1].MaxTXID != 13 || plan.MaxTXID != 13 {
+ t.Fatalf("selector output = %+v", plan)
+ }
+ path := filepath.Join(t.TempDir(), "restored.db")
+ if err := (ColdRestore{Reader: bucket, Bucket: "data", Generation: "new", Work: &restoreWork{}}).Install(ctx, path, plan); err != nil {
+ t.Fatalf("restore selector output: %v", err)
+ }
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var got string
+ if err := db.QueryRow(`SELECT value FROM fixture`).Scan(&got); err != nil || got != "compacted-at-13" {
+ t.Fatalf("restored value = %q, err = %v", got, err)
+ }
+ if string(mustRead(t, path+".gen")) != "new\n" {
+ t.Fatal("admitted generation marker missing")
+ }
+ })
+ }
+}
+
+func TestLoadPinnedRejectsGapAndNonadvancingRange(t *testing.T) {
+ for _, bounds := range [][2]ltx.TXID{{13, 13}, {10, 11}} {
+ t.Run(fmt.Sprint(bounds), func(t *testing.T) {
+ key, snapshot, _ := restoreLTXFixture(t, "snapshot", 9, 1, 11)
+ continuation, data, _ := restoreLTXFixture(t, "invalid", 1, bounds[0], bounds[1])
+ reader := &restoreReader{data: map[string][]byte{key + "@v1": snapshot, continuation + "@v2": data}}
+ plan := &genselect.Plan{Generation: "one", MaxTXID: uint64(bounds[1]), Files: []genselect.File{
+ {Key: key, VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: 11, Size: int64(len(snapshot))},
+ {Key: continuation, VersionID: "v2", Level: 1, MinTXID: uint64(bounds[0]), MaxTXID: uint64(bounds[1]), Size: int64(len(data))},
+ }}
+ if _, _, err := loadPinned(t.Context(), reader, "data", plan); !errors.Is(err, genselect.ErrGap) {
+ t.Fatalf("invalid continuation accepted: %v", err)
+ }
+ })
+ }
+}
+
+func snapshotFixture(t *testing.T, value string) (string, []byte, uint64) {
+ t.Helper()
+ return restoreLTXFixture(t, value, 9, 1, 1)
+}
+
+func restoreLTXFixture(t *testing.T, value string, level int, min, max ltx.TXID) (string, []byte, uint64) {
+ t.Helper()
+ path := filepath.Join(t.TempDir(), "fixture.db")
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = db.Exec(`CREATE TABLE fixture(value TEXT); INSERT INTO fixture VALUES (?)`, value); err != nil {
+ t.Fatal(err)
+ }
+ if err = db.Close(); err != nil {
+ t.Fatal(err)
+ }
+ raw := mustRead(t, path)
+ pageSize := int(binary.BigEndian.Uint16(raw[16:18]))
+ if pageSize == 1 {
+ pageSize = 65536
+ }
+ if len(raw)%pageSize != 0 {
+ t.Fatalf("database size %d not page aligned", len(raw))
+ }
+ spec := ltx.FileSpec{Header: ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: uint32(pageSize), Commit: uint32(len(raw) / pageSize), MinTXID: min, MaxTXID: max}}
+ for off := 0; off < len(raw); off += pageSize {
+ pgno := uint32(off/pageSize + 1)
+ if pgno == ltx.LockPgno(uint32(pageSize)) {
+ continue
+ }
+ spec.Pages = append(spec.Pages, ltx.PageSpec{Header: ltx.PageHeader{Pgno: pgno}, Data: append([]byte(nil), raw[off:off+pageSize]...)})
+ }
+ var out bytes.Buffer
+ if _, err := spec.WriteTo(&out); err != nil {
+ t.Fatal(err)
+ }
+ return fmt.Sprintf("instances/shop/g-one/%04x/%s", level, ltx.FormatFilename(min, max)), out.Bytes(), uint64(spec.Trailer.FileChecksum)
+}
+
+func mustRead(t *testing.T, path string) []byte {
+ t.Helper()
+ b, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
diff --git a/internal/hostbox/wakepolicy.go b/internal/hostbox/wakepolicy.go
new file mode 100644
index 000000000..9c3c00d65
--- /dev/null
+++ b/internal/hostbox/wakepolicy.go
@@ -0,0 +1,90 @@
+package hostbox
+
+import (
+ "context"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+// WakePhase is the K6 execution table. Participant membership alone never starts code.
+func WakePhase(c *hostclaim.Claim, box string, probe bool) error {
+ if c == nil || c.Holder == nil || box == "" || c.Op.Exec != box {
+ return ErrChanged
+ }
+ for _, lost := range c.Lost {
+ if lost == box {
+ return ErrChanged
+ }
+ }
+ switch c.Op.Kind {
+ case hostclaim.KindServing:
+ if !probe && c.Op.ID == "" && c.Op.Phase == "" {
+ return nil
+ }
+ case hostclaim.KindLanding:
+ if c.Op.ID != "" && ((!probe && c.Op.Phase == "committed") || (probe && c.Op.Phase == "probing")) {
+ return nil
+ }
+ case hostclaim.KindReverting, hostclaim.KindRestoring:
+ if c.Op.ID != "" && ((!probe && (c.Op.Phase == "probed" || c.Op.Phase == "opened")) || (probe && c.Op.Phase == "probing")) {
+ return nil
+ }
+ }
+ return ErrChanged
+}
+
+// Policy separates ordinary writes from the moves-owned held actions.
+// Moves validates roles, held inputs and phase-specific rights with the same capture.
+type Policy struct {
+ Inputs func(context.Context, *hostclaim.Claim, Selection, hostwrite.Identity) error
+ Moves func(context.Context, *hostclaim.Claim, Selection, hostwrite.Identity) error
+ Held func(context.Context, string) (string, error)
+}
+
+func (p Policy) Check(ctx context.Context, c *hostclaim.Claim, s Selection, id hostwrite.Identity) error {
+ if c == nil || p.Held == nil {
+ return ErrUnsupported
+ }
+ hold, err := p.Held(ctx, s.Host)
+ if err != nil {
+ return err
+ }
+ if c.Op.Kind == hostclaim.KindServing {
+ if id.Config != "" || id.Release != "" {
+ if p.Inputs == nil {
+ return ErrUnsupported
+ }
+ if err := p.Inputs(ctx, c, s, id); err != nil {
+ return err
+ }
+ }
+ if hold != "" {
+ return ErrChanged
+ }
+ return Serving(c, s)
+ }
+ if p.Moves == nil || s.Op == "" || c.Op.ID != s.Op || c.Op.Exec != s.Box {
+ return ErrChanged
+ }
+ switch s.Action {
+ case "held-config", "held-release", "held-timers":
+ if hold != s.Op {
+ return ErrChanged
+ }
+ case "restore", "spawn", "seed", "writer", "install":
+ if hold != "" {
+ if hold != s.Op {
+ return ErrChanged
+ }
+ } else if err := WakePhase(c, s.Box, false); err != nil {
+ return err
+ }
+ default:
+ return ErrChanged
+ }
+ return p.Moves(ctx, c, s, id)
+}
+
+const RequestLimit = 120 * time.Second
diff --git a/internal/hostbox/wakepolicy_test.go b/internal/hostbox/wakepolicy_test.go
new file mode 100644
index 000000000..f7dd9add5
--- /dev/null
+++ b/internal/hostbox/wakepolicy_test.go
@@ -0,0 +1,99 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/release"
+ "testing"
+)
+
+func TestWakePhaseK6(t *testing.T) {
+ for _, tc := range []struct {
+ kind, phase string
+ probe, want bool
+ }{
+ {"serving", "", false, true}, {"serving", "", true, false},
+ {"landing", "committed", false, true}, {"landing", "restored", false, false},
+ {"landing", "probing", true, true}, {"reverting", "probed", false, true},
+ {"restoring", "opened", false, true}, {"moving", "flipped", false, false},
+ } {
+ c := &hostclaim.Claim{Holder: &hostclaim.Holder{}, Op: hostclaim.Op{Kind: tc.kind, Phase: tc.phase, ID: "op", Exec: "box"}}
+ if tc.kind == "serving" {
+ c.Op.ID = ""
+ }
+ if got := WakePhase(c, "box", tc.probe) == nil; got != tc.want {
+ t.Fatalf("%+v: %v", tc, got)
+ }
+ if WakePhase(c, "other", tc.probe) == nil {
+ t.Fatal("foreign executor")
+ }
+ }
+}
+func TestHeldTimerPolicyOnlyOwningOperation(t *testing.T) {
+ c := &hostclaim.Claim{Holder: &hostclaim.Holder{}, Op: hostclaim.Op{Kind: "landing", ID: "landing", Exec: "box"}}
+ p := Policy{Held: func(context.Context, string) (string, error) { return "landing", nil }, Moves: func(context.Context, *hostclaim.Claim, Selection, hostwrite.Identity) error { return nil }}
+ s := Selection{Host: "x.example", Box: "box", Op: "landing", Action: "held-timers"}
+ if err := p.Check(context.Background(), c, s, hostwrite.Identity{}); err != nil {
+ t.Fatal(err)
+ }
+ for _, action := range []string{"timer-delivery", "config", "release"} {
+ s.Action = action
+ if p.Check(context.Background(), c, s, hostwrite.Identity{}) == nil {
+ t.Fatal(action)
+ }
+ }
+ s.Action = "held-timers"
+ s.Op = "previous"
+ if p.Check(context.Background(), c, s, hostwrite.Identity{}) == nil {
+ t.Fatal("stale op")
+ }
+}
+func TestWriterColdOnlyEvenWithCertificate(t *testing.T) {
+ evidenceCalls := 0
+ w := Writer{Claims: release.NewDirStore(t.TempDir()), Box: "box", Region: "ie", AdmissionEvidence: func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error {
+ evidenceCalls++
+ return nil
+ }}
+ obs := hostclaim.Observed{Host: "x.example", ETag: "one", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{}}}
+ for _, in := range []hostclaim.Admission{{Box: "box", Region: "ie", Mode: "warm", PredecessorCertified: true}, {Box: "box", Region: "ie", Mode: "cold", NewGeneration: "next", PredecessorCertified: true}} {
+ if _, err := w.Admit(context.Background(), obs, in, nil); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+ }
+ if evidenceCalls != 0 {
+ t.Fatal("certificate provider consulted", evidenceCalls)
+ }
+}
+
+func TestPolicyOrdinaryWakeAfterReleaseHold(t *testing.T) {
+ for _, phase := range []struct{ kind, phase string }{{"landing", "committed"}, {"reverting", "probed"}, {"reverting", "opened"}, {"restoring", "probed"}, {"restoring", "opened"}} {
+ c := &hostclaim.Claim{Holder: &hostclaim.Holder{}, Op: hostclaim.Op{Kind: phase.kind, Phase: phase.phase, ID: "op", Exec: "box"}}
+ p := Policy{Held: func(context.Context, string) (string, error) { return "", nil }, Moves: func(context.Context, *hostclaim.Claim, Selection, hostwrite.Identity) error { return nil }}
+ s := Selection{Host: "x.example", Box: "box", Op: "op"}
+ for _, action := range []string{"restore", "spawn", "seed", "writer", "install"} {
+ s.Action = action
+ if err := p.Check(t.Context(), c, s, hostwrite.Identity{}); err != nil {
+ t.Fatalf("%+v %s: %v", phase, action, err)
+ }
+ }
+ for _, action := range []string{"held-config", "held-release", "held-timers", "config", "timers"} {
+ s.Action = action
+ if p.Check(t.Context(), c, s, hostwrite.Identity{}) == nil {
+ t.Fatalf("unheld %s accepted", action)
+ }
+ }
+ s.Action = "spawn"
+ c.Op.Phase = "probing"
+ if p.Check(t.Context(), c, s, hostwrite.Identity{}) == nil {
+ t.Fatal("ordinary wake in probing")
+ }
+ c.Op.Phase = phase.phase
+ s.Op = "other"
+ if p.Check(t.Context(), c, s, hostwrite.Identity{}) == nil {
+ t.Fatal("stale op")
+ }
+ }
+}
diff --git a/internal/hostbox/writer.go b/internal/hostbox/writer.go
index 484b45cfd..db938995a 100644
--- a/internal/hostbox/writer.go
+++ b/internal/hostbox/writer.go
@@ -28,6 +28,9 @@ func (w Writer) Baseline(ctx context.Context, observed hostclaim.Observed) (*gen
}
func (w Writer) Admit(ctx context.Context, observed hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) (hostclaim.Observed, error) {
+ if in.Mode != "cold" || in.PredecessorCertified {
+ return hostclaim.Observed{}, ErrUnsupported
+ }
if w.Claims == nil || w.AdmissionEvidence == nil || w.Box == "" || w.Region == "" || observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil || in.Box != w.Box || in.Region != w.Region {
return hostclaim.Observed{}, ErrUnsupported
}
diff --git a/internal/lease/claimgeneration.go b/internal/lease/claimgeneration.go
new file mode 100644
index 000000000..a2f987404
--- /dev/null
+++ b/internal/lease/claimgeneration.go
@@ -0,0 +1,156 @@
+package lease
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "path"
+ "strings"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ "github.com/superfly/ltx"
+)
+
+var ErrGenerationRetention = errors.New("lease: generation retention requires exact versions through park")
+
+var ErrGenerationBinding = errors.New("lease: invalid or changed generation binding")
+
+const generationRequestLimit = 120 * time.Second
+
+// BindGeneration fixes all replica data requests below the held lease's
+// location. The authority callback is captured once and checked per request.
+func (h *Held) BindGeneration(prefix string, check func(context.Context) error) error {
+ if h == nil || h.leaser == nil || check == nil {
+ return ErrGenerationBinding
+ }
+ base := strings.TrimSuffix(path.Clean(h.leaser.Path), "/")
+ clean := path.Clean(prefix)
+ rel, ok := strings.CutPrefix(clean, base+"/")
+ if (prefix != clean && prefix != clean+"/") || strings.Contains(prefix, "\\") || base == "." || base == "" || !ok || strings.Contains(rel, "/") || !strings.HasPrefix(rel, "g-") || len(rel) == 2 {
+ return ErrGenerationBinding
+ }
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.busy || h.spent || h.generationPrefix != "" {
+ return ErrGenerationBinding
+ }
+ client := h.replicaClient
+ if client == nil {
+ client = newReplicaClient(h.leaser.Bucket, clean, h.region)
+ }
+ h.replicaClient = &generationClient{client: client, check: check}
+ h.generationPrefix = clean
+ return nil
+}
+
+type generationClient struct {
+ client litestream.ReplicaClient
+ check func(context.Context) error
+}
+
+func (c *generationClient) Type() string { return c.client.Type() }
+func (c *generationClient) SetLogger(l *slog.Logger) { c.client.SetLogger(l) }
+func generationLevel(level int) error {
+ if level < 0 || level > litestream.SnapshotLevel {
+ return ErrGenerationBinding
+ }
+ return nil
+}
+func generationCoords(level int, min, max ltx.TXID) error {
+ if generationLevel(level) != nil || min == 0 || max == 0 || min > max {
+ return ErrGenerationBinding
+ }
+ return nil
+}
+func (c *generationClient) bounded(ctx context.Context) (context.Context, context.CancelFunc, error) {
+ if ctx == nil {
+ return nil, nil, ErrGenerationBinding
+ }
+ ctx, cancel := context.WithTimeout(ctx, generationRequestLimit)
+ if err := c.check(ctx); err != nil {
+ cancel()
+ return nil, nil, err
+ }
+ return ctx, cancel, nil
+}
+func (c *generationClient) Init(ctx context.Context) error {
+ ctx, cancel, err := c.bounded(ctx)
+ if err != nil {
+ return err
+ }
+ defer cancel()
+ return c.client.Init(ctx)
+}
+func (c *generationClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, metadata bool) (ltx.FileIterator, error) {
+ if err := generationLevel(level); err != nil {
+ return nil, err
+ }
+ ctx, cancel, err := c.bounded(ctx)
+ if err != nil {
+ return nil, err
+ }
+ it, err := c.client.LTXFiles(ctx, level, seek, metadata)
+ if err != nil {
+ cancel()
+ return nil, err
+ }
+ return &generationIterator{FileIterator: it, cancel: cancel}, nil
+}
+func (c *generationClient) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ if err := generationCoords(level, min, max); err != nil || offset < 0 || size < 0 {
+ return nil, ErrGenerationBinding
+ }
+ ctx, cancel, err := c.bounded(ctx)
+ if err != nil {
+ return nil, err
+ }
+ rc, err := c.client.OpenLTXFile(ctx, level, min, max, offset, size)
+ if err != nil {
+ cancel()
+ return nil, err
+ }
+ return &generationReader{ReadCloser: rc, cancel: cancel}, nil
+}
+func (c *generationClient) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
+ if err := generationCoords(level, min, max); err != nil || r == nil {
+ return nil, ErrGenerationBinding
+ }
+ ctx, cancel, err := c.bounded(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer cancel()
+ return c.client.WriteLTXFile(ctx, level, min, max, r)
+}
+func (c *generationClient) DeleteLTXFiles(ctx context.Context, files []*ltx.FileInfo) error {
+ if len(files) == 0 {
+ return ErrGenerationBinding
+ }
+ for _, f := range files {
+ if f == nil || generationCoords(f.Level, f.MinTXID, f.MaxTXID) != nil {
+ return ErrGenerationBinding
+ }
+ }
+ // FileInfo has no VersionID. Only the park engine may retire exact versions.
+ return ErrGenerationRetention
+}
+func (c *generationClient) DeleteAll(context.Context) error {
+ return fmt.Errorf("%w: DeleteAll refused", ErrGenerationBinding)
+}
+
+type generationReader struct {
+ io.ReadCloser
+ cancel context.CancelFunc
+}
+
+func (r *generationReader) Close() error { err := r.ReadCloser.Close(); r.cancel(); return err }
+
+type generationIterator struct {
+ ltx.FileIterator
+ cancel context.CancelFunc
+}
+
+func (i *generationIterator) Close() error { err := i.FileIterator.Close(); i.cancel(); return err }
diff --git a/internal/lease/claimgeneration_external_test.go b/internal/lease/claimgeneration_external_test.go
new file mode 100644
index 000000000..83b8ecc55
--- /dev/null
+++ b/internal/lease/claimgeneration_external_test.go
@@ -0,0 +1,19 @@
+package lease_test
+
+import (
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "testing"
+)
+
+func TestBindGenerationAcceptsRealGenerationPrefix(t *testing.T) {
+ h := hostclaim.Holder{Location: "instances/x.example/inc", Generation: "one", Session: 1}
+ prefix, err := hostbox.GenerationPrefix(h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := lease.GenerationTestBinding(h.Location, prefix); err != nil {
+ t.Fatalf("real prefix %q: %v", prefix, err)
+ }
+}
diff --git a/internal/lease/claimgeneration_test.go b/internal/lease/claimgeneration_test.go
new file mode 100644
index 000000000..805fa7942
--- /dev/null
+++ b/internal/lease/claimgeneration_test.go
@@ -0,0 +1,196 @@
+package lease
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "io"
+ "log/slog"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ lss3 "github.com/benbjohnson/litestream/s3"
+ "github.com/superfly/ltx"
+)
+
+type generationFake struct {
+ calls int
+ ctx context.Context
+ files []*ltx.FileInfo
+}
+
+func (f *generationFake) Type() string { return "fake" }
+func (f *generationFake) SetLogger(*slog.Logger) {}
+func (f *generationFake) Init(ctx context.Context) error { f.calls++; f.ctx = ctx; return nil }
+func (f *generationFake) LTXFiles(ctx context.Context, _ int, _ ltx.TXID, _ bool) (ltx.FileIterator, error) {
+ f.calls++
+ f.ctx = ctx
+ return ltx.NewFileInfoSliceIterator(f.files), nil
+}
+func (f *generationFake) OpenLTXFile(ctx context.Context, _ int, _ ltx.TXID, _ ltx.TXID, _ int64, _ int64) (io.ReadCloser, error) {
+ f.calls++
+ f.ctx = ctx
+ return io.NopCloser(bytes.NewReader(nil)), nil
+}
+func (f *generationFake) WriteLTXFile(ctx context.Context, _ int, min, max ltx.TXID, _ io.Reader) (*ltx.FileInfo, error) {
+ f.calls++
+ f.ctx = ctx
+ return &ltx.FileInfo{MinTXID: min, MaxTXID: max}, nil
+}
+func (f *generationFake) DeleteLTXFiles(ctx context.Context, _ []*ltx.FileInfo) error {
+ f.calls++
+ f.ctx = ctx
+ return nil
+}
+func (f *generationFake) DeleteAll(context.Context) error { f.calls++; return nil }
+
+func TestBindGenerationIsImmutableAndCanonical(t *testing.T) {
+ leaser := lss3.NewLeaser()
+ leaser.Path = "instances/shop.example/inc"
+ f := &generationFake{}
+ h := &Held{leaser: leaser, replicaClient: f}
+ check := func(context.Context) error { return nil }
+ if err := h.BindGeneration("instances/shop.example/inc/g-generation1", check); err != nil {
+ t.Fatal(err)
+ }
+ if err := h.BindGeneration("instances/shop.example/inc/g-generation2", check); err != ErrGenerationBinding {
+ t.Fatalf("rebind=%v", err)
+ }
+ for _, bad := range []string{"instances/shop.example/inc/g-a/child", "instances/shop.example/inc/x/../g-a", "instances/shop.example/other/g-a", "instances/shop.example/inc/g-"} {
+ h2 := &Held{leaser: leaser, replicaClient: &generationFake{}}
+ if err := h2.BindGeneration(bad, check); err != ErrGenerationBinding {
+ t.Errorf("BindGeneration(%q)=%v", bad, err)
+ }
+ }
+}
+
+func TestGenerationClientValidatesBeforeStorage(t *testing.T) {
+ f := &generationFake{}
+ c := &generationClient{client: f, check: func(context.Context) error { return nil }}
+ if _, err := c.WriteLTXFile(t.Context(), -1, 1, 1, bytes.NewReader(nil)); err != ErrGenerationBinding {
+ t.Fatal(err)
+ }
+ if _, err := c.OpenLTXFile(t.Context(), 0, 2, 1, 0, 0); err != ErrGenerationBinding {
+ t.Fatal(err)
+ }
+ if err := c.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{nil}); err != ErrGenerationBinding {
+ t.Fatal(err)
+ }
+ if err := c.DeleteAll(t.Context()); err == nil {
+ t.Fatal("DeleteAll accepted")
+ }
+ if f.calls != 0 {
+ t.Fatalf("invalid requests reached storage: %d", f.calls)
+ }
+}
+
+func TestGenerationClientBoundsAndRetainsStreamContext(t *testing.T) {
+ checks := 0
+ f := &generationFake{}
+ c := &generationClient{client: f, check: func(context.Context) error { checks++; return nil }}
+ r, err := c.OpenLTXFile(t.Context(), litestream.SnapshotLevel, 1, 1, 0, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ deadline, ok := f.ctx.Deadline()
+ if !ok || time.Until(deadline) > generationRequestLimit || time.Until(deadline) < generationRequestLimit-time.Second {
+ t.Fatalf("deadline=%v ok=%v", deadline, ok)
+ }
+ select {
+ case <-f.ctx.Done():
+ t.Fatal("stream context canceled before Close")
+ default:
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case <-f.ctx.Done():
+ case <-time.After(time.Second):
+ t.Fatal("stream Close did not cancel request context")
+ }
+ if checks != 1 {
+ t.Fatalf("authority checks=%d", checks)
+ }
+}
+
+// GenerationTestBinding is a test-only bridge for the external package to
+// exercise hostbox.GenerationPrefix without a lease -> hostbox import cycle.
+func GenerationTestBinding(location, prefix string) error {
+ l := lss3.NewLeaser()
+ l.Path = location
+ h := &Held{leaser: l, replicaClient: &generationFake{}}
+ return h.BindGeneration(prefix, func(context.Context) error { return nil })
+}
+
+func TestGenerationClientRetentionRefusedButUploadsContinue(t *testing.T) {
+ f := &generationFake{}
+ c := &generationClient{client: f, check: func(context.Context) error { return nil }}
+ for _, level := range []int{0, 1, litestream.SnapshotLevel} {
+ if err := c.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{{Level: level, MinTXID: 1, MaxTXID: 2}}); !errors.Is(err, ErrGenerationRetention) {
+ t.Fatalf("level %d delete: %v", level, err)
+ }
+ }
+ if f.calls != 0 {
+ t.Fatal("bare-key deletion reached backing client")
+ }
+ if _, err := c.WriteLTXFile(t.Context(), 0, 3, 3, bytes.NewReader(nil)); err != nil {
+ t.Fatal(err)
+ }
+ if f.calls != 1 {
+ t.Fatal("replication stopped after refusal")
+ }
+}
+
+func TestGenerationClientDisablesRemoteRetentionInRealStore(t *testing.T) {
+ for _, bound := range []bool{false, true} {
+ l := lss3.NewLeaser()
+ l.Path = "instances/x.example/inc"
+ h := &Held{leaser: l, replicaClient: &generationFake{}}
+ if bound {
+ if err := h.BindGeneration(l.Path+"/g-one/", func(context.Context) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ _, db := h.newStore(filepath.Join(t.TempDir(), "db"), time.Second)
+ if db.RetentionEnabled == bound {
+ t.Fatalf("bound=%v retention=%v", bound, db.RetentionEnabled)
+ }
+ }
+}
+
+func TestGenerationClientRealRetentionMonitorsDoNotDelete(t *testing.T) {
+ l := lss3.NewLeaser()
+ l.Path = "instances/x.example/inc"
+ f := &generationFake{files: []*ltx.FileInfo{{Level: 1, MinTXID: 1, MaxTXID: 2, CreatedAt: time.Unix(1, 0)}, {Level: 1, MinTXID: 3, MaxTXID: 4, CreatedAt: time.Unix(2, 0)}}}
+ h := &Held{leaser: l, replicaClient: f}
+ if err := h.BindGeneration(l.Path+"/g-one/", func(context.Context) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ store, db := h.newStore(filepath.Join(t.TempDir(), "db"), time.Second)
+ if store.RetentionEnabled || db.RetentionEnabled {
+ t.Fatal("remote retention enabled")
+ }
+ db.MonitorInterval = 0
+ db.Replica.MonitorEnabled = false
+ if err := db.Open(); err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close(t.Context())
+ // These execute the real DB and compactor retention code. A deletion
+ // attempt would get ErrGenerationRetention and fail the call.
+ if err := db.EnforceRetentionByTXID(t.Context(), 1, 10); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.EnforceSnapshotRetention(t.Context(), time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if err := db.EnforceL0RetentionByTime(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Replica.Client.WriteLTXFile(t.Context(), 0, 5, 5, bytes.NewReader(nil)); err != nil {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/lease/lease.go b/internal/lease/lease.go
index 72761a941..effcd0cd9 100644
--- a/internal/lease/lease.go
+++ b/internal/lease/lease.go
@@ -100,14 +100,15 @@ func (e *HeldError) Error() string {
// file exists. Renewal runs from acquisition, so a slow restore can't silently
// outlive the TTL.
type Held struct {
- leaser *lss3.Leaser
- region string // the leaser's client keeps its region private; carry it for Replicate
- rn *renewer
- mu sync.Mutex
- busy bool
- spent bool
- poison error
- replicaClient litestream.ReplicaClient // private local-replica test seam
+ leaser *lss3.Leaser
+ region string // the leaser's client keeps its region private; carry it for Replicate
+ rn *renewer
+ mu sync.Mutex
+ busy bool
+ spent bool
+ poison error
+ replicaClient litestream.ReplicaClient // private local-replica test seam
+ generationPrefix string
}
// Acquire claims the single-writer lease at s3://bucket/prefix/lock.json
@@ -295,6 +296,11 @@ func (h *Held) newStore(dbPath string, syncInterval time.Duration) (*litestream.
db.Replica.SyncInterval = syncInterval
}
store := litestream.NewStore([]*litestream.DB{db}, compactionLevels)
+ if h.generationPrefix != "" {
+ // NewStore resets the DB flag. Set all three store/DB/compactor flags
+ // afterward, before Open starts any monitor or compaction work.
+ store.SetRetentionEnabled(false)
+ }
store.L0RetentionCheckInterval = l0RetentionCheckInterval
return store, db
}
diff --git a/internal/moves/install_contract.go b/internal/moves/install_contract.go
new file mode 100644
index 000000000..f48d8fe76
--- /dev/null
+++ b/internal/moves/install_contract.go
@@ -0,0 +1,22 @@
+package moves
+
+import (
+ "errors"
+ "time"
+)
+
+var ErrAttemptOver = errors.New("moves: probe attempt window or deadline expired")
+
+// InstallManifest is the moves Phase 1 Plan 2 Task 8 prepared snapshot contract.
+type InstallManifest struct {
+ V int
+ Op, Host string
+ N uint64
+ SnapshotKey, SnapshotVersion string
+ SnapshotCRC64 uint64
+ AnchorKey, AnchorVersion string
+ AnchorCRC64 uint64
+ IntegrityCheck string
+ DBSHA256 string
+ BuiltAt time.Time
+}
diff --git a/internal/provision/moveinputs_contract.go b/internal/provision/moveinputs_contract.go
new file mode 100644
index 000000000..fd51aa09e
--- /dev/null
+++ b/internal/provision/moveinputs_contract.go
@@ -0,0 +1,10 @@
+package provision
+
+// MoveInputs pins the inputs checked by the moves owner before admission.
+type MoveInputs struct {
+ BundleETags map[string]string `json:"bundle_etags,omitempty"`
+ OverlayETag string `json:"overlay_etag,omitempty"`
+ Release string `json:"release,omitempty"`
+ FilesPlain string `json:"files_plain,omitempty"`
+ EnvReport string `json:"env_report,omitempty"`
+}
diff --git a/internal/schedule/claimreplace_test.go b/internal/schedule/claimreplace_test.go
new file mode 100644
index 000000000..8fb4ee224
--- /dev/null
+++ b/internal/schedule/claimreplace_test.go
@@ -0,0 +1,43 @@
+package schedule
+
+import (
+ "context"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+ "path/filepath"
+ "testing"
+)
+
+type replacementWork struct{ valid *bool }
+
+func (w replacementWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
+ if !*w.valid {
+ return hostwrite.ErrUnavailable
+ }
+ return fn()
+}
+func TestSelectedHeldReplacementKeepsRegistrationAndRefusesStale(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ valid := true
+ selected := SelectedTimers{Store: reg, Work: replacementWork{&valid}, Context: context.Background(), Host: "shop.example"}
+ rows := []registry.Timer{{Host: "shop.example", Name: "one", Kind: "once", Spec: "spec", Path: "/tick", NextAt: 123, DueAt: 120, Registration: "from-source"}}
+ if err := selected.ReplaceTimers("shop.example", rows); err != nil {
+ t.Fatal(err)
+ }
+ got, err := reg.GetTimer("shop.example", "one")
+ if err != nil || got.Registration != "from-source" || got.DueAt != 120 {
+ t.Fatal(got, err)
+ }
+ valid = false
+ if err := selected.ReplaceTimers("shop.example", nil); err == nil {
+ t.Fatal("stale landing cleared timers")
+ }
+ got, err = reg.GetTimer("shop.example", "one")
+ if err != nil || got.Registration != "from-source" {
+ t.Fatal(got, err)
+ }
+}
diff --git a/internal/schedule/claimtimers.go b/internal/schedule/claimtimers.go
index ff2b208d2..abee98d44 100644
--- a/internal/schedule/claimtimers.go
+++ b/internal/schedule/claimtimers.go
@@ -61,3 +61,21 @@ func (s SelectedTimers) CountTimers(host, kind string) (int, error) {
}
var _ TimerStore = SelectedTimers{}
+
+// ReplaceTimers imports a landing snapshot without changing registrations.
+// Its Work must have been selected as held-timers for the owning landing op.
+func (s SelectedTimers) ReplaceTimers(host string, rows []registry.Timer) error {
+ replacement, ok := s.Store.(interface {
+ ReplaceTimers(string, []registry.Timer) error
+ })
+ if !ok {
+ return hostwrite.ErrUnavailable
+ }
+ copyRows := append([]registry.Timer(nil), rows...)
+ for _, row := range copyRows {
+ if row.Host != host {
+ return errTimerHost
+ }
+ }
+ return s.write(host, func() error { return replacement.ReplaceTimers(host, copyRows) })
+}