amadan
host: the exported end-to-end bridge, E2E
NewE2E(t, store, opts...) builds the two-box world over the caller's deployment store, with both boxes on the self-serve switch, a lane to each other's data bucket, an armed stage path and an instance id. E2E drives them as the agent does: Pass (reconcile, edges file, moves pass on every box that is up), Serial (passes until the serial is consumed), Status (the self-serve status block and the box's snapshot), Clock (the boxes' clock), Fault (pass:, status:, copy: and seal: points per box) and Box (a BoxHandle with the box id, instance id, registry, data bucket and data directory, ArmOffline, Reconcile, Pass, Tick, Status, Down and Up). It also hands a console built beside it the secrets and release keys.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 file changed,
+370
−0
internal/host/e2e_bridge.go+370 −0
diff --git a/internal/host/e2e_bridge.go b/internal/host/e2e_bridge.go| index c8e3614a6..28a32db0c 100644 |
| --- a/internal/host/e2e_bridge.go |
| +++ b/internal/host/e2e_bridge.go |
| @@ -919,3 +919,373 @@ func (lw *landWorld) tickAll() { |
| lw.tick(lw.ie) |
| lw.tick(lw.uk) |
| } |
| + |
| +// --- the exported bridge ------------------------------------------------------ |
| + |
| +// The bridge's deployment: one instance, E2EHost of app "app", at home on |
| +// E2ESourceBox, with E2ETargetBox the one box of E2ETargetRegion. A console |
| +// built beside the bridge takes E2EHomeRegion as its HomeRegion. |
| +const ( |
| + E2EHost = offHost |
| + E2EApp = "app" |
| + E2ESourceBox = "ie-box" // the home's one box, data bucket data-ie |
| + E2ETargetBox = "uk-box" // uk's one box, data bucket data-uk |
| + E2EHomeRegion = "ie" |
| + E2ETargetRegion = "uk" |
| +) |
| + |
| +// The fault points the bridge reaches. A point is one of these prefixes and |
| +// a box label (FaultPass + E2ESourceBox), and its function runs on the |
| +// test's goroutine at that point, before the step it names. |
| +const ( |
| + // FaultPass: before the box's tick (reconcile, edges file, moves pass) |
| + // in Pass, Serial and BoxHandle.Tick. |
| + FaultPass = "pass:" |
| + // 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 = "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 = "seal:" |
| +) |
| + |
| +var e2eFaultPoints = []string{FaultPass, FaultStatus, FaultCopy, FaultSeal} |
| + |
| +// E2E is the two-box offline move world for end-to-end witnesses outside |
| +// this package (internal/e2e/selfserve): the boxes of this package's own |
| +// offline move tests (newTwoBoxWorld), driven through the functions the |
| +// agent calls (newMovesWorker armed by armOffline, reconcileSerializedWith, |
| +// the production edges-file writer, movesPass and the status tick), over |
| +// the deployment store the caller's console uses. |
| +// |
| +// NewE2E sets process environment (t.Setenv), so a test using it cannot |
| +// run in parallel. The boxes run on the test's goroutine, when the test |
| +// calls Pass, Serial or Status; nothing runs in the background. |
| +type E2E struct { |
| + t testing.TB |
| + lw *landWorld |
| + keys e2eKeys |
| + clock *Clock |
| + boxes []*e2eBox |
| + mu sync.Mutex |
| + faults map[string]func() |
| +} |
| + |
| +type e2eBox struct { |
| + b *offBox |
| + down atomic.Bool |
| +} |
| + |
| +// E2EOption shapes the bridge's deployment. |
| +type E2EOption func(*e2eSetup) |
| + |
| +type e2eSetup struct { |
| + acct e2eAccount |
| + opts landOpts |
| +} |
| + |
| +// E2EAccount makes the account with this sqid, console id and name own the |
| +// app (its manifest is written only when the store has none). The default |
| +// is account 2, "fixture". |
| +func E2EAccount(sqid string, id int64, name string) E2EOption { |
| + return func(s *e2eSetup) { s.acct = e2eAccount{sqid: sqid, id: id, name: name} } |
| +} |
| + |
| +// E2EJudged arms both boxes with a lifetime judge, as a claims-off agent |
| +// arms newBoxJudge's. |
| +func E2EJudged() E2EOption { return func(s *e2eSetup) { s.opts.judged = true } } |
| + |
| +// E2ELifetime is E2EJudged with the record created through its lifetime: |
| +// the host's incarnation slot at 1, live, then the record at incarnation 1. |
| +func E2ELifetime() E2EOption { return func(s *e2eSetup) { s.opts.lifetime = true } } |
| + |
| +// NewE2E builds the bridge over store, the deployment store the console |
| +// under test uses: the account's and app's manifests, the instance's record |
| +// at home, two shipped versions (the first promoted) and the app's config |
| +// bundle; both boxes armed, with CARLOS_SELF_SERVE_MOVES=offline, a lane to |
| +// each other's data bucket and an armed stage path; the source at rest with |
| +// a database replicated to data-ie, a files directory and an env file; and |
| +// both boxes published once. The data buckets are served over the S3 wire |
| +// protocol on loopback for the test's lifetime. |
| +func NewE2E(t testing.TB, store release.Store, opts ...E2EOption) *E2E { |
| + t.Helper() |
| + setup := e2eSetup{acct: e2eAccount{sqid: domains.Sqid(2), id: 2, name: "fixture"}} |
| + for _, o := range opts { |
| + o(&setup) |
| + } |
| + 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} { |
| + lane := sealUK |
| + if b == lw.uk { |
| + lane = sealIE |
| + } |
| + b.env[selfServeMovesEnv] = "offline" |
| + b.env[moveLanesEnv] = lane |
| + id := "i-e2e-" + b.name |
| + b.instanceID.Store(&id) |
| + b.stagePath = func() *bool { return &armed } |
| + x := &e2eBox{b: b} |
| + e.boxes = append(e.boxes, x) |
| + e.rearm(x) |
| + } |
| + }) |
| + e.clock = &Clock{lw: e.lw} |
| + 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) { |
| + 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) { |
| + e.fault(FaultCopy + label) |
| + return mirror(ctx, dst) |
| + } |
| + } |
| + if carry := c.CarryRequest; carry != nil { |
| + c.CarryRequest = func(ctx context.Context, host, bucket, prefix string) error { |
| + e.fault(FaultSeal + label) |
| + return carry(ctx, host, bucket, prefix) |
| + } |
| + } |
| +} |
| + |
| +// Store is the deployment store the bridge was built over. |
| +func (e *E2E) Store() release.Store { return e.lw.wd.s } |
| + |
| +// Account is the sqid of the account that owns E2EApp. |
| +func (e *E2E) Account() string { return e.lw.wd.acct } |
| + |
| +// SecretsPub is the deployment secrets key's public half, as |
| +// console.Config.SecretsPub takes it; both boxes hold the private half. |
| +func (e *E2E) SecretsPub() string { return e.keys.secretsPub() } |
| + |
| +// ReleasePub and ReleaseKey are the release key's halves, as |
| +// console.Config.ReleasePub and Key take them; the target box verifies |
| +// with the public half. |
| +func (e *E2E) ReleasePub() ed25519.PublicKey { return e.keys.relPub } |
| +func (e *E2E) ReleaseKey() ed25519.PrivateKey { return e.keys.relPriv } |
| + |
| +// Clock is the boxes' clock. |
| +func (e *E2E) Clock() *Clock { return e.clock } |
| + |
| +// Fault runs f at point (one of the Fault prefixes and a box label) from |
| +// now on, replacing any function already there; a nil f clears the point. |
| +// A point that is not one of the bridge's fails the test. |
| +func (e *E2E) Fault(point string, f func()) { |
| + e.t.Helper() |
| + known := false |
| + for _, prefix := range e2eFaultPoints { |
| + if label, ok := strings.CutPrefix(point, prefix); ok && e.find(label) != nil { |
| + known = true |
| + } |
| + } |
| + if !known { |
| + e.t.Fatalf("e2e: no fault point %q", point) |
| + } |
| + e.mu.Lock() |
| + defer e.mu.Unlock() |
| + if f == nil { |
| + delete(e.faults, point) |
| + return |
| + } |
| + e.faults[point] = f |
| +} |
| + |
| +func (e *E2E) fault(point string) { |
| + e.mu.Lock() |
| + f := e.faults[point] |
| + e.mu.Unlock() |
| + if f != nil { |
| + f() |
| + } |
| +} |
| + |
| +func (e *E2E) find(label string) *e2eBox { |
| + for _, x := range e.boxes { |
| + if x.b.name == label { |
| + return x |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// Box is the box with this label (E2ESourceBox or E2ETargetBox). |
| +func (e *E2E) Box(label string) BoxHandle { |
| + e.t.Helper() |
| + x := e.find(label) |
| + if x == nil { |
| + e.t.Fatalf("e2e: no box %q", label) |
| + } |
| + return BoxHandle{e: e, x: x} |
| +} |
| + |
| +// Pass is one agent tick on every box that is up, source first: the |
| +// reconcile pass, the edges file and the moves pass, in the agent's order. |
| +// Each step runs whatever the one before it returned, as in the agent; the |
| +// error is every step's, joined. |
| +func (e *E2E) Pass(ctx context.Context) error { |
| + var errs []error |
| + for _, x := range e.boxes { |
| + if !x.down.Load() { |
| + errs = append(errs, e.tick(ctx, x)) |
| + } |
| + } |
| + return errors.Join(errs...) |
| +} |
| + |
| +// e2eSerialPasses bounds Serial. |
| +const e2eSerialPasses = 16 |
| + |
| +// Serial runs Pass until the deployment's serial (release.SerialKey) is |
| +// consumed: a Pass that leaves it as it found it, as an agent that has |
| +// caught up with every bump runs no further pass. |
| +func (e *E2E) Serial(ctx context.Context) error { |
| + for range e2eSerialPasses { |
| + _, before, err := release.ReadSerial(ctx, e.lw.wd.s) |
| + if err != nil { |
| + return err |
| + } |
| + if err := e.Pass(ctx); err != nil { |
| + return err |
| + } |
| + _, after, err := release.ReadSerial(ctx, e.lw.wd.s) |
| + if err != nil { |
| + return err |
| + } |
| + if after == before { |
| + return nil |
| + } |
| + } |
| + return fmt.Errorf("e2e: the serial was still moving after %d passes", e2eSerialPasses) |
| +} |
| + |
| +// Status is the status tick on every box that is up: the self-serve block |
| +// (selfServeStatusProbes) and the box's snapshot (statusTick, never the |
| +// fleet publication). |
| +func (e *E2E) Status(ctx context.Context) error { |
| + var errs []error |
| + for _, x := range e.boxes { |
| + if !x.down.Load() { |
| + errs = append(errs, e.status(ctx, x)) |
| + } |
| + } |
| + return errors.Join(errs...) |
| +} |
| + |
| +func (e *E2E) tick(ctx context.Context, x *e2eBox) error { |
| + b := x.b |
| + e.fault(FaultPass + b.name) |
| + var errs []error |
| + if err := b.reconcileErr(ctx); err != nil { |
| + errs = append(errs, fmt.Errorf("%s: reconcile: %w", b.name, err)) |
| + } |
| + if err := e.lw.publishErr(ctx, b); err != nil { |
| + errs = append(errs, fmt.Errorf("%s: edges file: %w", b.name, err)) |
| + } |
| + if err := movesPass(ctx, e.lw.wd.s, b.w); err != nil { |
| + errs = append(errs, fmt.Errorf("%s: moves pass: %w", b.name, err)) |
| + } |
| + return errors.Join(errs...) |
| +} |
| + |
| +func (e *E2E) status(ctx context.Context, x *e2eBox) error { |
| + b := x.b |
| + 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 { |
| + 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 } |
| + |
| +// 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) } |
| + |
| +// Advance moves the boxes' time on by d. |
| +func (c *Clock) Advance(d time.Duration) { c.lw.wd.advance(d) } |
| + |
| +// BoxHandle is one of the bridge's boxes. |
| +type BoxHandle struct { |
| + e *E2E |
| + x *e2eBox |
| +} |
| + |
| +// Label is the box's label, as its edges file and status snapshot name it. |
| +func (h BoxHandle) Label() string { return h.x.b.name } |
| + |
| +// BoxID is the box's id, as its moves worker and reports name it |
| +// (moveCfg.BoxID). |
| +func (h BoxHandle) BoxID() string { return h.x.b.w.cfg.BoxID } |
| + |
| +// InstanceID is the box's cloud instance id, as its moves worker and |
| +// status block report it (moveCfg.instance). |
| +func (h BoxHandle) InstanceID() string { return h.x.b.w.cfg.instance() } |
| + |
| +// SetInstanceID makes the box report another cloud instance id from now |
| +// on: a box replaced under the same label. |
| +func (h BoxHandle) SetInstanceID(id string) { h.x.b.instanceID.Store(&id) } |
| + |
| +// Region is the box's placement ("" is the home). |
| +func (h BoxHandle) Region() string { return h.x.b.region } |
| + |
| +// Registry is the box's own registry. |
| +func (h BoxHandle) Registry() *registry.Registry { return h.x.b.reg } |
| + |
| +// DataBucket is the box's data bucket (CARLOS_HIBERNATE_BUCKET). |
| +func (h BoxHandle) DataBucket() string { return h.x.b.getenv("CARLOS_HIBERNATE_BUCKET") } |
| + |
| +// DataDir is the box's data directory. |
| +func (h BoxHandle) DataDir() string { return h.x.b.dd } |
| + |
| +// ArmOffline is an agent restart: a fresh activator (no fences) and a |
| +// fresh moves worker armed by armOffline over the same registry, data |
| +// directory and buckets, as the box was first armed. |
| +func (h BoxHandle) ArmOffline() { h.e.rearm(h.x) } |
| + |
| +// Reconcile is the box's reconcile pass (reconcileSerializedWith over its |
| +// worker). |
| +func (h BoxHandle) Reconcile(ctx context.Context) error { return h.x.b.reconcileErr(ctx) } |
| + |
| +// Pass is the box's moves pass alone (movesPass). |
| +func (h BoxHandle) Pass(ctx context.Context) error { |
| + return movesPass(ctx, h.e.lw.wd.s, h.x.b.w) |
| +} |
| + |
| +// Tick is one agent tick on this box, as Pass runs it on every box. |
| +func (h BoxHandle) Tick(ctx context.Context) error { return h.e.tick(ctx, h.x) } |
| + |
| +// Status is the box's status tick, as E2E.Status runs it. |
| +func (h BoxHandle) Status(ctx context.Context) error { return h.e.status(ctx, h.x) } |
| + |
| +// Down takes the box away: Pass, Serial and Status skip it until Up, as |
| +// they would a box that stopped. Its registry, disk and buckets stay. |
| +func (h BoxHandle) Down() { h.x.down.Store(true) } |
| + |
| +// Up brings the box back. |
| +func (h BoxHandle) Up() { h.x.down.Store(false) } |