amadan

carlos / platform Public

host: the bridge's copy point per round, a clock that follows the wall until set, knobs

copy: now rides the stager's round-start seam and the sealer's final point
(the worker's mirror is cached per region, so the Mirror wrapper fired once
per region, most often on a report write). The bridge clock follows the
wall clock at every pass and status tick until the test first sets or
advances it, then drives every box-side time, the status snapshot included
(statusTickAt). BoxHandle gains SetStagePathArmed and SetDiskFree.
sealPrefix is its own const declaration, so the census names it stably.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude faff9c7724826fcf7d37b3f2d71c774e077bf8aa parent 42e263d
1 file changed, +132 −29
  • internal/host/e2e_bridge.go +132 −29
diff --git a/internal/host/e2e_bridge.go b/internal/host/e2e_bridge.go
index 4f929535f..a3273e267 100644
--- a/internal/host/e2e_bridge.go
+++ b/internal/host/e2e_bridge.go
@@ -19,6 +19,11 @@ package host
// exported bridge (E2E, NewE2E, BoxHandle, Clock and the fault points) is
// at the end of the file. Everything here is portable: the offline move
// code it reaches has no Linux-only piece, so no _linux.go split is needed.
+//
+// The bridge's fault points ride the worker's own test seams: copy: on the
+// stager's "round-start" (moveCfg.stageFault) and the sealer's "final"
+// (offlineSealFaults), seal: on every moveCfg.CarryRequest call, whoever
+// makes it (in an offline move, only the sealer). See FaultCopy.
import (
"context"
@@ -102,12 +107,14 @@ type offBox struct {
w *movesWorker
nested bool
last []byte
- // The bridge's identity hooks (E2E): the cloud instance id the box
- // reports, read on every use as moveCfg.InstanceIDOf, and the stage
- // path's answer. Both unset in this package's own worlds, which then
- // arm exactly as before: an empty id, and armOffline's stagePathActive.
+ // The bridge's hooks (E2E): the cloud instance id the box reports,
+ // read on every use as moveCfg.InstanceIDOf, the stage path's answer
+ // (moveCfg.StagePathArmed) and the disk's (moveCfg.DiskFree). All unset
+ // in this package's own worlds, which then arm exactly as before: an
+ // empty id, and armOffline's stagePathActive and diskBytes.
instanceID atomic.Pointer[string]
stagePath func() *bool
+ diskFree func(dir string) (total, avail uint64, err error)
}
// instanceIDOf is moveCfg.InstanceIDOf for the box: "" until the bridge
@@ -147,7 +154,7 @@ func (b *offBox) getenv(k string) string {
func (b *offBox) restart() {
b.svc = activator.New(activator.Config{Bucket: "data", Idle: time.Hour}, &fakeWakeBackend{}, b.reg)
cfg := moveCfg{Judge: b.wd.judge, Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock,
- InstanceIDOf: b.instanceIDOf, StagePathArmed: b.stagePath}
+ InstanceIDOf: b.instanceIDOf, StagePathArmed: b.stagePath, DiskFree: b.diskFree}
armOffline(&cfg, b.dd, b.reg, b.svc, nil, hostwrite.Config{}, b.getenv)
// The stage-requested marker is root's business, and its throttled
// timer would outlive the test's directories.
@@ -224,10 +231,12 @@ const (
sealUK = "data-uk"
sealIERegion = "eu-west-1"
sealUKRegion = "eu-west-2"
- sealPrefix = "instances"
sealVersion = "v42"
)
+// sealPrefix is the data buckets' hibernate prefix, never a record key.
+const sealPrefix = "instances"
+
type sealWorld struct {
*offWorld
fake *leasetest.VersionedBucket
@@ -429,7 +438,7 @@ func (lw *landWorld) arm(b *offBox) {
bucket, region := b.getenv("CARLOS_HIBERNATE_BUCKET"), b.getenv("CARLOS_HIBERNATE_REGION")
b.svc = activator.New(activator.Config{Bucket: bucket, Idle: time.Hour}, &fakeWakeBackend{}, b.reg)
cfg := moveCfg{Judge: b.wd.judge, Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock,
- InstanceIDOf: b.instanceIDOf, StagePathArmed: b.stagePath,
+ InstanceIDOf: b.instanceIDOf, StagePathArmed: b.stagePath, DiskFree: b.diskFree,
OwnBucket: bucket, OwnBucketRegion: region, OwnPrefix: sealPrefix,
Mirror: func(ctx context.Context, dst string) (lease.MirrorStore, error) {
return lease.NewS3Mirror(ctx, region, dst)
@@ -948,13 +957,19 @@ const (
// FaultStatus: before the box's status tick in Status and
// BoxHandle.Status.
FaultStatus = "status:"
- // FaultCopy: each time the box's moves worker opens a mirror into
- // another bucket (moveCfg.Mirror): the stager's rounds, the sealer's
- // and the lander's copies.
+ // FaultCopy: as each copy round between buckets begins, before any
+ // object is copied. It fires once per stager round, at the round's
+ // start (moveCfg.stageFault "round-start", after the stager has found a
+ // stage request it has not yet answered), and once per sealer attempt
+ // at the final round (the sealer's "final" point, after the tombstone
+ // and before the build state's carry and the byte-final copy). The
+ // lander copies nothing between buckets (it reads its own), so it
+ // never fires there.
FaultCopy = "copy:"
- // FaultSeal: inside the sealer's step on the source, under its fence,
- // as it carries the build state before the final copy round
- // (moveCfg.CarryRequest): a deadline that passes mid-seal.
+ // FaultSeal: every call of the box's moveCfg.CarryRequest. In an
+ // offline move only the sealer calls it, on the source, under its
+ // fence, after FaultCopy's "final" point and before the byte-final copy
+ // round: a deadline that passes mid-seal.
FaultSeal = "seal:"
)
@@ -983,6 +998,32 @@ type E2E struct {
type e2eBox struct {
b *offBox
down atomic.Bool
+ // The knobs (BoxHandle.SetStagePathArmed, SetDiskFree), read on every
+ // use: the stage path's answer (nil: unknown) and the disk's (nil: the
+ // box's real statfs, diskBytes).
+ knobs sync.Mutex
+ stage *bool
+ disk func(dir string) (total, avail uint64, err error)
+}
+
+func (x *e2eBox) stagePathArmed() *bool {
+ x.knobs.Lock()
+ defer x.knobs.Unlock()
+ if x.stage == nil {
+ return nil
+ }
+ v := *x.stage
+ return &v
+}
+
+func (x *e2eBox) diskFree(dir string) (total, avail uint64, err error) {
+ x.knobs.Lock()
+ f := x.disk
+ x.knobs.Unlock()
+ if f == nil {
+ return diskBytes(dir)
+ }
+ return f(dir)
}
// E2EOption shapes the bridge's deployment.
@@ -1024,7 +1065,6 @@ func NewE2E(t testing.TB, store release.Store, opts ...E2EOption) *E2E {
}
e := &E2E{t: t, keys: newE2EKeys(t), faults: map[string]func(){}}
fake := newE2EEnv(t)
- armed := true
newTwoBoxWorld(t, store, fake, e.keys, setup.acct, time.Now(), setup.opts, func(lw *landWorld) {
e.lw = lw
for _, b := range []*offBox{lw.ie, lw.uk} {
@@ -1036,28 +1076,43 @@ func NewE2E(t testing.TB, store release.Store, opts ...E2EOption) *E2E {
b.env[moveLanesEnv] = lane
id := "i-e2e-" + b.name
b.instanceID.Store(&id)
- b.stagePath = func() *bool { return &armed }
- x := &e2eBox{b: b}
+ armed := true
+ x := &e2eBox{b: b, stage: &armed}
+ b.stagePath, b.diskFree = x.stagePathArmed, x.diskFree
e.boxes = append(e.boxes, x)
e.rearm(x)
}
})
e.clock = &Clock{lw: e.lw}
+ t.Cleanup(func() {
+ for _, x := range e.boxes {
+ offlineSealFaults.Delete(x.b.w)
+ }
+ })
return e
}
// rearm is an agent restart of x's box: armOffline over a fresh activator
// and worker, armed as the world first armed it, then the fault points.
func (e *E2E) rearm(x *e2eBox) {
+ if x.b.w != nil {
+ offlineSealFaults.Delete(x.b.w)
+ }
e.lw.rearm[x.b]()
label := x.b.name
c := &x.b.w.cfg
- if mirror := c.Mirror; mirror != nil {
- c.Mirror = func(ctx context.Context, dst string) (lease.MirrorStore, error) {
+ c.stageFault = func(point string) error {
+ if point == "round-start" {
e.fault(FaultCopy + label)
- return mirror(ctx, dst)
}
+ return nil
}
+ offlineSealFaults.Store(x.b.w, func(point string) error {
+ if point == "final" {
+ e.fault(FaultCopy + label)
+ }
+ return nil
+ })
if carry := c.CarryRequest; carry != nil {
c.CarryRequest = func(ctx context.Context, host, bucket, prefix string) error {
e.fault(FaultSeal + label)
@@ -1191,6 +1246,7 @@ func (e *E2E) Status(ctx context.Context) error {
func (e *E2E) tick(ctx context.Context, x *e2eBox) error {
b := x.b
+ e.clock.follow()
e.fault(FaultPass + b.name)
var errs []error
if err := b.reconcileErr(ctx); err != nil {
@@ -1207,32 +1263,55 @@ func (e *E2E) tick(ctx context.Context, x *e2eBox) error {
func (e *E2E) status(ctx context.Context, x *e2eBox) error {
b := x.b
+ now := e.clock.follow()
e.fault(FaultStatus + b.name)
routes, err := b.reg.List()
if err != nil {
return fmt.Errorf("%s: status tick: %w", b.name, err)
}
p := status.Probes{MachineID: b.name}
- selfServeStatusProbes(ctx, &p, b.w, b.w.cfg.BoxID, b.getenv, routes, e.clock.Now())
- if err := statusTick(ctx, e.lw.wd.s, b.reg, p, false, b.name); err != nil {
+ selfServeStatusProbes(ctx, &p, b.w, b.w.cfg.BoxID, b.getenv, routes, now)
+ if err := statusTickAt(ctx, e.lw.wd.s, b.reg, p, false, b.name, now); err != nil {
return fmt.Errorf("%s: status tick: %w", b.name, err)
}
return nil
}
-// Clock is the boxes' clock: every box's moveCfg.Now reads it. NewE2E
-// leaves it at the wall clock's reading of the boxes' first publication;
-// it moves only when the test moves it.
-type Clock struct{ lw *landWorld }
+// Clock is the boxes' clock: every box-side time the bridge drives reads
+// it (moveCfg.Now, so the moves pass and every report's stamps, and the
+// status tick's snapshot, statusTickAt). Until the test first calls Set or
+// Advance it follows the wall clock: each Pass, Serial round, Tick and
+// status tick first sets it to time.Now(). From the first Set or Advance
+// on it moves only when the test moves it.
+type Clock struct {
+ lw *landWorld
+ manual atomic.Bool
+}
// Now is the boxes' time.
func (c *Clock) Now() time.Time { return c.lw.wd.clock() }
-// Set sets the boxes' time.
-func (c *Clock) Set(at time.Time) { c.lw.setBoxClock(at) }
+// Set sets the boxes' time, and stops it following the wall clock.
+func (c *Clock) Set(at time.Time) {
+ c.manual.Store(true)
+ c.lw.setBoxClock(at)
+}
-// Advance moves the boxes' time on by d.
-func (c *Clock) Advance(d time.Duration) { c.lw.wd.advance(d) }
+// Advance moves the boxes' time on by d, and stops it following the wall
+// clock.
+func (c *Clock) Advance(d time.Duration) {
+ c.manual.Store(true)
+ c.lw.wd.advance(d)
+}
+
+// follow is the clock's reading for a box-side step: the wall clock's,
+// until the test first sets or advances it.
+func (c *Clock) follow() time.Time {
+ if !c.manual.Load() {
+ c.lw.setBoxClock(time.Now())
+ }
+ return c.Now()
+}
// BoxHandle is one of the bridge's boxes.
type BoxHandle struct {
@@ -1278,6 +1357,7 @@ func (h BoxHandle) Reconcile(ctx context.Context) error { return h.x.b.reconcile
// Pass is the box's moves pass alone (movesPass).
func (h BoxHandle) Pass(ctx context.Context) error {
+ h.e.clock.follow()
return movesPass(ctx, h.e.lw.wd.s, h.x.b.w)
}
@@ -1293,3 +1373,26 @@ func (h BoxHandle) Down() { h.x.down.Store(true) }
// Up brings the box back.
func (h BoxHandle) Up() { h.x.down.Store(false) }
+
+// SetStagePathArmed sets the box's answer to whether root's stage path is
+// active (moveCfg.StagePathArmed, the precheck's and the ready report's
+// fact): nil is "cannot tell". NewE2E arms it (true).
+func (h BoxHandle) SetStagePathArmed(armed *bool) {
+ h.x.knobs.Lock()
+ defer h.x.knobs.Unlock()
+ if armed == nil {
+ h.x.stage = nil
+ return
+ }
+ v := *armed
+ h.x.stage = &v
+}
+
+// SetDiskFree sets the box's disk answer (moveCfg.DiskFree: total and
+// available bytes of the filesystem holding dir). nil restores the box's
+// real statfs of its temporary directory.
+func (h BoxHandle) SetDiskFree(f func(dir string) (total, avail uint64, err error)) {
+ h.x.knobs.Lock()
+ defer h.x.knobs.Unlock()
+ h.x.disk = f
+}