amadan

carlos / platform Public

host: root serves one held-release request per host and three in all (Task 7b fix round 2, P3-4)

Each stage can download for up to 15 minutes inside instance-control,
which also serves the claim's stop, retire and journal routes, and the
agent could open any number. Root now admits one request per host and
heldReleaseConcurrency (3) across hosts, refusing anything over either
bound at once with errHeldReleaseBusy (503) before any read; the agent's
client reads 503 as unavailable and retries on a later pass.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5d06e596bd8a9c14bde30e695223820385a972ff parent 09ac47b
3 files changed, +126 −1
  • internal/host/movesheldrelease.go +7 −0
  • internal/host/movesheldrelease_root.go +32 −1
  • internal/host/movesheldrelease_test.go +87 −0
diff --git a/internal/host/movesheldrelease.go b/internal/host/movesheldrelease.go
index 24afc6b72..229aea475 100644
--- a/internal/host/movesheldrelease.go
+++ b/internal/host/movesheldrelease.go
@@ -64,6 +64,9 @@ const (
// heldReleaseDeadline bounds one call. A stage may download the build,
// as T7 does in the agent with the claim off.
heldReleaseDeadline = 15 * time.Minute
+ // heldReleaseConcurrency bounds the held-release requests root serves at
+ // once across hosts (fix round 2, P3-4); each host has one at a time.
+ heldReleaseConcurrency = 3
)
var (
@@ -80,6 +83,10 @@ var (
// retry runs the whole path (the same render, the same selection checks)
// and publishes the same release.
errHeldReleaseReceipt = errors.New("moves: held release: published, but root's receipt did not persist")
+ // errHeldReleaseBusy is root at its bound (fix round 2, P3-4): a request
+ // for the host is already in flight, or heldReleaseConcurrency are. It
+ // answers 503, which the agent reads as unavailable and retries.
+ errHeldReleaseBusy = errors.New("moves: held release: root is busy with another request")
)
// heldReleaseRequest is one request. Kind, Phase, Incarnation, Stamp,
diff --git a/internal/host/movesheldrelease_root.go b/internal/host/movesheldrelease_root.go
index 390871a20..c780c2a62 100644
--- a/internal/host/movesheldrelease_root.go
+++ b/internal/host/movesheldrelease_root.go
@@ -13,6 +13,7 @@ import (
"os"
"path/filepath"
"slices"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/hostbox"
@@ -71,6 +72,31 @@ type movesHeldReleaseRoot struct {
materialize func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate, hostwrite.Work, func(func() error) error) (*release.Manifest, error)
renderHash func(dataDir, host, op string) (string, error)
afterPublish func(host string) error
+
+ // The bound (fix round 2, P3-4): one request per host at a time, and
+ // heldReleaseConcurrency in all.
+ mu sync.Mutex
+ inflight map[string]bool
+}
+
+// admit takes host's slot and one of the global ones, or refuses at once
+// with errHeldReleaseBusy: root never queues an agent's requests. The
+// returned func frees both.
+func (h *movesHeldReleaseRoot) admit(host string) (func(), error) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if h.inflight == nil {
+ h.inflight = map[string]bool{}
+ }
+ if h.inflight[host] || len(h.inflight) >= heldReleaseConcurrency {
+ return nil, errHeldReleaseBusy
+ }
+ h.inflight[host] = true
+ return func() {
+ h.mu.Lock()
+ delete(h.inflight, host)
+ h.mu.Unlock()
+ }, nil
}
func (h *movesHeldReleaseRoot) complete() bool {
@@ -131,7 +157,7 @@ func heldReleaseStatus(err error) int {
return http.StatusTooEarly
case errors.Is(err, release.ErrHeldNotProven):
return http.StatusUnprocessableEntity
- case errors.Is(err, errHeldReleaseUnavailable), errors.Is(err, hostwrite.ErrUnavailable):
+ case errors.Is(err, errHeldReleaseUnavailable), errors.Is(err, errHeldReleaseBusy), errors.Is(err, hostwrite.ErrUnavailable):
return http.StatusServiceUnavailable
case errors.Is(err, release.ErrStaleHeldStage), errors.Is(err, hostbox.ErrChanged), errors.Is(err, errHeldReleaseUnsafe),
errors.Is(err, release.ErrHeldNotAuthorized), errors.Is(err, release.ErrManifestDigest), errors.As(err, &refusal):
@@ -153,6 +179,11 @@ func (h *movesHeldReleaseRoot) Serve(ctx context.Context, peer StopPeer, q heldR
if err := q.validate(); err != nil {
return heldReleaseResponse{}, err
}
+ done, err := h.admit(q.Host)
+ if err != nil {
+ return heldReleaseResponse{}, err
+ }
+ defer done()
actor, err := h.Actor(ctx)
if err != nil {
return heldReleaseResponse{}, err
diff --git a/internal/host/movesheldrelease_test.go b/internal/host/movesheldrelease_test.go
index 1975b4bd0..712bbf879 100644
--- a/internal/host/movesheldrelease_test.go
+++ b/internal/host/movesheldrelease_test.go
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"os"
"path/filepath"
"strings"
@@ -804,3 +805,89 @@ func TestHeldReleaseRootRefusesUnjournalledRecoveryProbes(t *testing.T) {
})
}
}
+
+// TestHeldReleaseRootBoundsConcurrency (fix round 2, P3-4): root serves
+// one request per host at a time, and at most heldReleaseConcurrency in
+// all, so the agent cannot pile up downloads inside instance-control. A
+// request over either bound is refused at once as busy (503, which the
+// agent's client reads as unavailable and retries), before any read; a
+// slot is free again once its request returns.
+func TestHeldReleaseRootBoundsConcurrency(t *testing.T) {
+ ctx := context.Background()
+ hold := func(f *hrFixture) (entered chan string, release func()) {
+ entered = make(chan string, 16)
+ gate := make(chan struct{})
+ claims := f.root.Claims
+ f.root.Claims = func(ctx context.Context, host string) (hostclaim.Observed, error) {
+ entered <- host
+ <-gate
+ return claims(ctx, host)
+ }
+ var once sync.Once
+ release = func() { once.Do(func() { close(gate) }) }
+ t.Cleanup(release)
+ return entered, release
+ }
+ busy := func(t *testing.T, err error) {
+ t.Helper()
+ if !errors.Is(err, errHeldReleaseBusy) {
+ t.Fatalf("over the bound: %v, want errHeldReleaseBusy", err)
+ }
+ if got := heldReleaseStatus(err); got != 503 {
+ t.Fatalf("busy answers %d, want 503", got)
+ }
+ }
+
+ t.Run("one per host", func(t *testing.T) {
+ f := newHRFixture(t)
+ entered, release := hold(f)
+ done := make(chan error, 1)
+ go func() {
+ _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
+ done <- err
+ }()
+ <-entered
+ reads := f.claimReads
+ _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleaseStage))
+ busy(t, err)
+ if f.claimReads != reads || len(entered) != 0 {
+ t.Fatal("a busy request read the claim")
+ }
+ release()
+ if err := <-done; err != nil {
+ t.Fatalf("the first request: %v", err)
+ }
+ if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil {
+ t.Fatalf("after the first returned: %v", err)
+ }
+ })
+
+ t.Run("a bound across hosts", func(t *testing.T) {
+ f := newHRFixture(t)
+ entered, release := hold(f)
+ other := func(i int) heldReleaseRequest {
+ q := f.request(heldReleaseStage)
+ q.Host = fmt.Sprintf("h%d.go.example.com", i)
+ return q
+ }
+ done := make(chan error, heldReleaseConcurrency)
+ for i := 0; i < heldReleaseConcurrency; i++ {
+ go func(i int) {
+ _, err := f.root.Serve(ctx, hrPeer, other(i))
+ done <- err
+ }(i)
+ <-entered
+ }
+ _, err := f.root.Serve(ctx, hrPeer, other(99))
+ busy(t, err)
+ release()
+ for i := 0; i < heldReleaseConcurrency; i++ {
+ if err := <-done; errors.Is(err, errHeldReleaseBusy) {
+ t.Fatalf("a held request: %v", err)
+ }
+ }
+ if _, err := f.root.Serve(ctx, hrPeer, other(99)); errors.Is(err, errHeldReleaseBusy) {
+ t.Fatalf("after the slots freed: %v", err)
+ }
+ })
+}