amadan

carlos / platform Public

host: the local edge subscribes to the armed offline view's holds and takes the box id

Self-serve moves plan 3 Task 2, the agent wiring.

- subscribeEdgeHolds (the edge's ready callback) and
  subscribeArmedHolds (after armMovesPass arms the worker) subscribe
  edge.SetHoldView to the armed view under holdSubMu, whichever comes
  second, once per view; the view's Subscribe delivers its current
  snapshot in the same critical section.
- runAgent hands the edge edge.WithBoxID with the label the moves
  worker is armed with (moveCfg.BoxID).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude d37469eaef538ce4db3310c7afc457bd81176b5e parent 8fcd14c
4 files changed, +196 −0
  • internal/host/agent.go +9 −0
  • internal/host/edgeholds_test.go +139 −0
  • internal/host/movesoffline.go +44 −0
  • internal/host/movespass.go +4 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 9e2dd6985..30f8496d7 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -899,6 +899,10 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// the two derivations must be one derivation. Without it the edge
// would offer nothing (fail-closed), never dial itself.
edge.WithBoxLabel(metricsBoxLabel(os.Getenv, readMachineID())),
+ // This box's id, the value the moves worker is armed with
+ // (moveCfg.BoxID, armMovesPass: the same label), for the served-by
+ // stamp of a probe the edge answers here (self-serve moves plan 3).
+ edge.WithBoxID(metricsBoxLabel(os.Getenv, readMachineID())),
},
// The edge-local door (#175 P7), only when CARLOS_EDGE_DOOR names a
// socket. It presents this box's public address, the one the edges
@@ -911,6 +915,11 @@ func runAgent(args []string, buildClaims ClaimFactory) {
}
edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
go tunnelLoop(e)
+ // The offline view's holds reach this edge (self-serve moves plan
+ // 3): the holding page for this box's own held rows. The moves
+ // worker may arm before or after this; subscribeArmedHolds covers
+ // the other order.
+ subscribeEdgeHolds(e.SetHoldView)
// The edge owns the health probers, so its state table is only
// reachable once it exists — statusLoop (already running above) reads
// it through this published source on its next tick. Method value, so
diff --git a/internal/host/edgeholds_test.go b/internal/host/edgeholds_test.go
new file mode 100644
index 000000000..79587035b
--- /dev/null
+++ b/internal/host/edgeholds_test.go
@@ -0,0 +1,139 @@
+package host
+
+// The agent's wiring of the offline view into the local edge (self-serve
+// moves plan 3 Task 2): the view's holds reach edge.SetHoldView whichever of
+// the two arms first, a publication before the subscription is delivered at
+// it, and one racing it is delivered once, in revision order.
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// heldEdge is a real edge over a registry holding host under move.
+func heldEdge(t *testing.T, host, move string) (*edge.Edge, *registry.Registry) {
+ t.Helper()
+ reg := edgesTestRegistry(t)
+ heldInstance(t, reg, host, "moving", move, 1, 1)
+ e, err := edge.New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return e, reg
+}
+
+func servesPage(e *edge.Edge, host string) bool {
+ r := httptest.NewRequest(http.MethodGet, "https://"+host+"/", nil)
+ r.Host = host
+ w := httptest.NewRecorder()
+ e.ServeHTTP(w, r)
+ return w.Code == http.StatusServiceUnavailable && strings.Contains(w.Body.String(), "temporarily offline")
+}
+
+// armView makes v the armed worker's offline view for the test.
+func armView(t *testing.T, v *offlineView) {
+ t.Helper()
+ prev := armedMovesWorker.Load()
+ t.Cleanup(func() { armedMovesWorker.Store(prev) })
+ armedMovesWorker.Store(&movesWorker{cfg: moveCfg{OfflineCapable: true, Offline: v}})
+}
+
+// resetHoldSubscription clears the wiring's state for the test.
+func resetHoldSubscription(t *testing.T) {
+ t.Helper()
+ holdSubMu.Lock()
+ holdEdge, holdSubView = nil, nil
+ holdSubMu.Unlock()
+ t.Cleanup(func() {
+ holdSubMu.Lock()
+ holdEdge, holdSubView = nil, nil
+ holdSubMu.Unlock()
+ })
+}
+
+func TestLocalHoldsFollowTheViewAndTheRegistry(t *testing.T) {
+ const host = "cart.bab.example.net"
+ t.Run("a publication before the edge subscribes is delivered at subscription", func(t *testing.T) {
+ resetHoldSubscription(t)
+ e, _ := heldEdge(t, host, holdMoveA)
+ v := viewOf(map[string]provision.MoveState{host: selfServeMove(holdMoveA, provision.OfflineSealing)})
+ armView(t, v)
+ if servesPage(e, host) || e.Allowed(host) {
+ t.Fatal("test bug: nothing is subscribed yet")
+ }
+ subscribeEdgeHolds(e.SetHoldView)
+ if !servesPage(e, host) || !e.Allowed(host) {
+ t.Fatal("the view's earlier publication must reach the edge at subscription")
+ }
+ })
+ t.Run("the edge first, the worker after", func(t *testing.T) {
+ resetHoldSubscription(t)
+ e, _ := heldEdge(t, host, holdMoveA)
+ prev := armedMovesWorker.Load()
+ t.Cleanup(func() { armedMovesWorker.Store(prev) })
+ armedMovesWorker.Store(nil)
+ subscribeEdgeHolds(e.SetHoldView)
+ v := viewOf(map[string]provision.MoveState{host: selfServeMove(holdMoveA, provision.OfflineSealing)})
+ armedMovesWorker.Store(&movesWorker{cfg: moveCfg{OfflineCapable: true, Offline: v}})
+ subscribeArmedHolds()
+ if !servesPage(e, host) {
+ t.Fatal("a worker armed after the edge must subscribe it")
+ }
+ // A later view change reaches the edge through the subscription.
+ v.refresh(nil, nil)
+ if servesPage(e, host) {
+ t.Fatal("a cleared move must drop the hold")
+ }
+ })
+ t.Run("publications racing the subscription are delivered once, in order", func(t *testing.T) {
+ resetHoldSubscription(t)
+ e, _ := heldEdge(t, host, holdMoveA)
+ v := newOfflineView(moveCfg{}, nil)
+ armView(t, v)
+ var mu sync.Mutex
+ var revs []uint64
+ record := func(s edge.HoldSnapshot) {
+ mu.Lock()
+ revs = append(revs, s.Rev)
+ mu.Unlock()
+ e.SetHoldView(s)
+ }
+ mv := selfServeMove(holdMoveA, provision.OfflineSealing)
+ recs := []provision.Record{{Host: host, Account: "bab", App: "shop", Move: &mv}}
+ var wg sync.WaitGroup
+ for i := 0; i < 4; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for j := 0; j < 25; j++ {
+ v.refresh(recs, nil)
+ }
+ }()
+ }
+ subscribeEdgeHolds(record)
+ wg.Wait()
+ mu.Lock()
+ defer mu.Unlock()
+ if len(revs) == 0 {
+ t.Fatal("nothing was delivered")
+ }
+ for i := 1; i < len(revs); i++ {
+ if revs[i] != revs[i-1]+1 {
+ t.Fatalf("revisions %v: each delivered once, in order, two publishers never sharing one", revs)
+ }
+ }
+ if last := v.snapshot().Rev; revs[len(revs)-1] != last {
+ t.Fatalf("last delivered %d, last published %d", revs[len(revs)-1], last)
+ }
+ if !servesPage(e, host) {
+ t.Fatal("the hold must be installed")
+ }
+ })
+}
diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go
index 1065f9e76..afe3a5517 100644
--- a/internal/host/movesoffline.go
+++ b/internal/host/movesoffline.go
@@ -509,6 +509,50 @@ func armedOfflineView() *offlineView {
return w.cfg.Offline
}
+// The local edge's subscription to the armed offline view (self-serve moves
+// plan 3): the view's holds reach edge.SetHoldView, so this box's own edge
+// answers its held rows and their explicit aliases with the holding page.
+// The edge is ready (subscribeEdgeHolds) and the moves worker armed
+// (subscribeArmedHolds) in either order, on two goroutines; whichever comes
+// second subscribes, under holdSubMu, once per view. holdSubMu is taken
+// before the view's pubMu and never inside it (the edge's callback takes
+// no lock of this file's).
+var (
+ holdSubMu sync.Mutex
+ holdEdge func(edge.HoldSnapshot)
+ holdSubView *offlineView
+)
+
+// subscribeEdgeHolds records the ready edge's callback and subscribes it to
+// the armed view, when one is armed.
+func subscribeEdgeHolds(fn func(edge.HoldSnapshot)) {
+ holdSubMu.Lock()
+ defer holdSubMu.Unlock()
+ holdEdge = fn
+ subscribeHoldsLocked()
+}
+
+// subscribeArmedHolds subscribes the ready edge, when there is one, to the
+// view the moves worker just armed.
+func subscribeArmedHolds() {
+ holdSubMu.Lock()
+ defer holdSubMu.Unlock()
+ subscribeHoldsLocked()
+}
+
+// subscribeHoldsLocked is the one subscription: the view's Subscribe
+// delivers its current snapshot inside its own publication lock, so a
+// publication before it is seen and one racing it is delivered exactly once,
+// in revision order.
+func subscribeHoldsLocked() {
+ v := armedOfflineView()
+ if holdEdge == nil || v == nil || v == holdSubView {
+ return
+ }
+ v.Subscribe(holdEdge)
+ holdSubView = v
+}
+
// armedOfflineCapable is whether this box publishes offline-move-1: the
// armed worker is capable and has its keep view, exactly the condition
// under which reconcileSerializedWith installs the offline hooks.
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 1ee44482c..6c51cded7 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -366,6 +366,10 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
armOffline(&cfg, dataDir, reg, svc, eb, writes, os.Getenv)
w.cfg = cfg
armedMovesWorker.Store(w)
+ // The local edge answers this box's held rows from the view's holds
+ // (self-serve moves plan 3), whether it became ready before this or
+ // becomes ready after.
+ subscribeArmedHolds()
}
// armOffline sets every moveCfg field the offline operator move reads