F-catalog-B: distribution counts incoming against its binding's cap and drops an idle transfer by its expiring key

The holder counted against 20 GiB while the bus caps incoming at 8 GiB, so the bus refused writes before
the holder said refused naming the cap, and the 80 % finding came too late (running review F2). The cap is
read from incoming's binding (the object store's fill); IncomingCap, 8 GiB, is said only where none is read,
and the manifest's row holds it equal to max-mib. An abandoned transfer was dropped only on the next publish,
and the transfers state's own lifetime dropped its key while its parts stayed: transfers keeps for good, and
each open transfer has a key in the new state idle with its own time to live, whose end drops it (re-armed
for the rest when it took a part since). registry_incoming reports the bound cap.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/distribution/cmd/distribution-tools/main.go

@@ -15,10 +15,12 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/busstate"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/health"
 	"git.novox.be/novox/mesh-sdk/go/parts"
@@ -164,6 +166,53 @@ func learnInto(state casState, module, fp string, attempts int) (bool, error) {
 	return false, fmt.Errorf("%s was written by another writer first %d times: %w", holder.PublishingKey, attempts, stdio.ErrChanged)
 }
 
+// incomingCap reads incoming's cap from its binding (the bus's bound of the object store, design 51 *Publishing*), once
+// it answers one: the binding's cap does not change while the bundle runs.
+func incomingCap() func() (uint64, error) {
+	var known atomic.Uint64
+	return func() (uint64, error) {
+		if c := known.Load(); c > 0 {
+			return c, nil
+		}
+		f, err := busstate.ObjectsOf(holder.IncomingObjects).Fill()
+		if err != nil {
+			return 0, fmt.Errorf("incoming's cap cannot be read from its binding: %w", err)
+		}
+		if f.MaxBytes == 0 {
+			return 0, errors.New("incoming's binding carries no cap")
+		}
+		known.Store(f.MaxBytes)
+		return f.MaxBytes, nil
+	}
+}
+
+// armIdle writes a transfer's idle key with its own time to live; one already there is the same bound, kept.
+func armIdle(id string, ttl time.Duration) error {
+	if _, err := stdio.State(holder.IdleState).CreateTTL(id, map[string]any{"transfer": id}, ttl); err != nil && !errors.Is(err, stdio.ErrChanged) {
+		return err
+	}
+	return nil
+}
+
+// followIdle drops a transfer when its idle key ends (holder.Expired): the bus's marker is the event, no timer runs.
+func followIdle(h *holder.Holder) {
+	<-stdio.Served()
+	stdio.State(holder.IdleState).Follow("", func(ch stdio.StateChange) error {
+		if ch.Op != stdio.OpExpire {
+			return nil
+		}
+		dropped, err := h.Expired(ch.Key)
+		if err != nil {
+			logf("%v", err)
+			return err
+		}
+		if dropped {
+			logf("transfer %s had no part for %s: dropped with its parts", ch.Key, holder.TransferIdle)
+		}
+		return nil
+	})
+}
+
 // mirrors is the state mirrors: an ask is created once; one already kept is the same copy, taken on already.
 type mirrors struct{}
 
@@ -245,7 +294,6 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
 					if err != nil {
 						return nil, err
 					}
-					_, _ = v.DropIdle()
 					a, err := v.Publish(seat, stdio.CallerOf(args), publishOf(in))
 					if err == nil {
 						learnPublishing(windows, stdio.CallerOf(args))
@@ -397,7 +445,7 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
 				}
 				all, err := v.Transfers.List()
 				used, _ := v.Objects.Used()
-				return map[string]any{"transfers": all, "used": used, "cap": holder.IncomingCap, "dropped_idle": idle}, err
+				return map[string]any{"transfers": all, "used": used, "cap": v.CapNow(), "dropped_idle": idle}, err
 			}},
 		stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).",
 			Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}),
@@ -451,9 +499,10 @@ func main() {
 	base, _ := url.Parse("http://127.0.0.1:5001")
 	store := &holder.OCI{Base: base, Upstream: &holder.Upstream{}}
 	h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding,
-		Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState)}
+		Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState), Cap: incomingCap(), Idle: armIdle}
 	v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}}
 	go followMirrors(v)
+	go followIdle(h)
 	windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }}
 	if err := stdio.Serve("", tools(v, windows, holder.ConfiguredCallers)); err != nil {
 		logf("%v", err)

modules/distribution/internal/holder/holder_test.go

@@ -194,6 +194,9 @@ func newBed(builderOK bool) *bed {
 	return b
 }
 
+// capOf is a binding's cap of n bytes.
+func capOf(n uint64) func() (uint64, error) { return func() (uint64, error) { return n, nil } }
+
 func keys[V any](m map[string]V) []string {
 	var out []string
 	for k := range m {
@@ -267,7 +270,11 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
 				_, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t", Digest: part.Digest, Data: make([]byte, MaxPart+1)})
 				return err
 			}, "over the"},
-			{"a new transfer at the cap", func(b *bed) error { b.h.Cap = 100; _, err := b.h.Publish(ArtifactSeat, builder, req("t")); return err }, "at their cap"},
+			{"a new transfer at the cap", func(b *bed) error {
+				b.h.Cap = capOf(100)
+				_, err := b.h.Publish(ArtifactSeat, builder, req("t"))
+				return err
+			}, "at their cap"},
 			{"a version the builder did not number", func(b *bed) error {
 				b.h.Builder = fakeBuilder{ok: false}
 				b.store.has[part.Digest] = true // held already: the publish ends at once
@@ -332,10 +339,38 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
 		if dropped, _ := b.h.DropIdle(); !reflect.DeepEqual(dropped, []string{"t2"}) {
 			t.Errorf("DropIdle = %v, want t2 after 24 hours with no part", dropped)
 		}
+		// Running review F2, ADR 0611 §3: the drop follows each transfer's idle key. The open arms it for 24 hours; its
+		// end re-arms a transfer that took a part since for the rest, and drops one idle 24 hours, its chunks with it.
+		b = newBed(true)
+		armed := map[string]time.Duration{}
+		b.h.Idle = func(id string, ttl time.Duration) error { armed[id] = ttl; return nil }
+		if _, err := b.h.Publish(ArtifactSeat, builder, req("t4")); err != nil || armed["t4"] != TransferIdle {
+			t.Fatalf("publish = %v, armed %v; want t4's idle key for %s", err, armed, TransferIdle)
+		}
+		b.now = b.now.Add(time.Hour)
+		_, _ = b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t4", Digest: part.Digest, Data: blob[:100]})
+		b.now = b.now.Add(TransferIdle - time.Hour)
+		if dropped, err := b.h.Expired("t4"); err != nil || dropped || armed["t4"] != time.Hour {
+			t.Errorf("the key's end an hour after a part: dropped %v, %v, armed %v; want t4 kept and armed for 1h", dropped, err, armed)
+		}
+		b.now = b.now.Add(time.Hour)
+		if dropped, err := b.h.Expired("t4"); err != nil || !dropped {
+			t.Errorf("the key's end 24 hours after the last part: dropped %v, %v; want t4 dropped", dropped, err)
+		}
+		if _, found, _ := b.h.Transfers.Get("t4"); found || len(b.h.Objects.(memObjects)) != 0 {
+			t.Errorf("t4 after its drop: kept %v, objects %v; want both gone", found, b.h.Objects)
+		}
+		if dropped, err := b.h.Expired("t4"); err != nil || dropped {
+			t.Errorf("the end of a key whose transfer is gone: %v, %v; want nothing done", dropped, err)
+		}
+		b.h.Idle = func(string, time.Duration) error { return errors.New("bus down") }
+		if _, err := b.h.Publish(ArtifactSeat, builder, req("t5")); err == nil || !strings.Contains(err.Error(), "idle bound") {
+			t.Errorf("a publish whose idle key is not written = %v; want it failed loudly", err)
+		}
 	})
 	t.Run("incoming's finding at 80 %, cleared below", func(t *testing.T) {
 		b := newBed(true)
-		b.h.Cap = 3500
+		b.h.Cap = capOf(3500)
 		_, _ = b.h.Publish(ArtifactSeat, builder, req("t3"))
 		_, _ = b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t3", Digest: part.Digest, Data: blob[:2800]})
 		if !b.findings["incoming-full incoming"] {
@@ -346,12 +381,12 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
 			t.Error("the finding stayed after the transfer was dropped")
 		}
 		// A refusal for want of room stays loud on incoming until the holder takes a publish again.
-		b.h.Cap = 100
+		b.h.Cap = capOf(100)
 		_, _ = b.h.Publish(ArtifactSeat, builder, req("t4"))
 		if !b.findings["publish-refused incoming"] {
 			t.Errorf("a refusal at the cap raised %v, want publish-refused on incoming", b.findings)
 		}
-		b.h.Cap = 1 << 20
+		b.h.Cap = capOf(1 << 20)
 		b.store.has[part.Digest] = true
 		if _, err := b.h.Publish(ArtifactSeat, builder, req("t5")); err != nil || b.findings["publish-refused incoming"] {
 			t.Errorf("the next publish taken = %v, findings %v: want the refusal at the cap cleared", err, b.findings)
@@ -764,5 +799,28 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
 		if store < 0 || program < 0 || store > program {
 			t.Errorf("store at %d, program at %d: want the store listed first", store, program)
 		}
+		// Running review F2: incoming asks the bus 8 GiB, the most the objects template allows, equal to IncomingCap
+		// (said only where no binding is read); transfers keeps for good, the holder dropping an idle one by its key
+		// in idle, a state of keys with their own time to live.
+		var req struct {
+			Requires []struct {
+				Name    string         `json:"name"`
+				Options map[string]any `json:"options"`
+			} `json:"requires"`
+		}
+		_ = json.Unmarshal(raw, &req)
+		opts := map[string]map[string]any{}
+		for _, r := range req.Requires {
+			opts[r.Name] = r.Options
+		}
+		if mib, _ := opts[IncomingObjects]["max-mib"].(float64); uint64(mib)<<20 != IncomingCap {
+			t.Errorf("incoming asks %v MiB, IncomingCap is %d bytes: want them equal", opts[IncomingObjects]["max-mib"], uint64(IncomingCap))
+		}
+		if _, ok := opts[TransfersState]["ttl-seconds"]; ok {
+			t.Errorf("transfers has a lifetime %v: the bus would drop a transfer's key and leave its parts", opts[TransfersState])
+		}
+		if opts[IdleState]["key-ttl"] != true {
+			t.Errorf("idle's options %v: want key-ttl, each transfer's key ending on its own", opts[IdleState])
+		}
 	})
 }

modules/distribution/internal/holder/transfer.go

@@ -12,16 +12,22 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/parts"
 )
 
-// The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, the store 20 GiB, a transfer
-// with no part for 24 hours is dropped, and past 80 % of the cap the holder writes a finding.
+// The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, a transfer with no part for 24
+// hours is dropped, and past 80 % of the cap the holder writes a finding. The cap is the one incoming's binding carries
+// (the bus's own bound of the object store, read through the runtime); IncomingCap is the manifest's 8 GiB, the most
+// the objects template allows, said only where no binding is read (a test), and the manifest's test holds the two equal.
 const (
 	MaxPart         = 1 << 20
-	IncomingCap     = 20 << 30
+	IncomingCap     = 8 << 30
 	TransferIdle    = 24 * time.Hour
 	IncomingFullPct = 80
 	// IncomingObjects and TransfersState are the object store and the state distribution asks as bus state.
 	IncomingObjects = "incoming"
 	TransfersState  = "transfers"
+	// IdleState is the state of one key per open transfer with its own time to live (a state declared key-ttl): the
+	// bus's marker of its end is the event that drops a transfer left with no part for TransferIdle (ADR 0611 §3), never
+	// a timer.
+	IdleState = "idle"
 )
 
 // The kinds a seat holds.
@@ -156,8 +162,10 @@ type Holder struct {
 	Say       Say
 	Finding   Finding
 	Now       func() time.Time
-	// Cap is incoming's cap; IncomingCap when zero.
-	Cap uint64
+	// Cap reads incoming's cap from its binding; IncomingCap when nil or when the binding names none.
+	Cap func() (uint64, error)
+	// Idle writes a transfer's idle key with its own time to live, once; one already there is no fault. Nil arms nothing.
+	Idle func(id string, ttl time.Duration) error
 }
 
 func (h *Holder) now() time.Time {
@@ -167,13 +175,18 @@ func (h *Holder) now() time.Time {
 	return time.Now().UTC()
 }
 
-func (h *Holder) cap() uint64 {
-	if h.Cap > 0 {
-		return h.Cap
+// CapNow is incoming's cap as its binding carries it.
+func (h *Holder) CapNow() uint64 {
+	if h.Cap != nil {
+		if c, err := h.Cap(); err == nil && c > 0 {
+			return c
+		}
 	}
 	return IncomingCap
 }
 
+func (h *Holder) cap() uint64 { return h.CapNow() }
+
 // receiver binds mesh-sdk's receiving side to distribution's incoming, its transfers state, the builder's check, its
 // store and its seats' events.
 func (h *Holder) receiver() *parts.Receiver[Opening] {
@@ -255,6 +268,13 @@ func (h *Holder) Publish(seat, caller string, r PublishRequest) (PublishAnswer,
 	if err != nil {
 		return answerOf(got), fmt.Errorf("publish: %w", err)
 	}
+	// An open transfer is bounded by its idle key; a key that cannot be written fails the publish loudly, and the
+	// build module's next publish of the same transfer answers the parts held and arms it again.
+	if !got.Ended && h.Idle != nil {
+		if err := h.Idle(r.Transfer, TransferIdle); err != nil {
+			return answerOf(got), fmt.Errorf("publish: transfer %s is open but its idle bound is not kept, so it would never be dropped: %w", r.Transfer, err)
+		}
+	}
 	return answerOf(got), nil
 }
 
@@ -283,10 +303,34 @@ func (h *Holder) eventOf(seat string, r PublishRequest, ref, reason string) Even
 // Drop drops a transfer by hand (registry_incoming) or because it was idle for TransferIdle: its chunks and its key.
 func (h *Holder) Drop(id string) (Transfer, error) { return h.receiver().Drop(id) }
 
-// DropIdle drops every transfer with no part for TransferIdle, answering them; it runs on each publish's open and each
-// registry_incoming call, never on a timer (the object store's own lifetime of 24 hours drops the bytes too).
+// DropIdle drops every transfer with no part for TransferIdle, answering them: registry_incoming's look, by hand. The
+// drop itself follows each transfer's idle key (Expired); the objects template keeps for good, so nothing else does.
 func (h *Holder) DropIdle() ([]string, error) { return h.receiver().DropIdle() }
 
+// Expired is told that a transfer's idle key ended (the bus's marker): a transfer with no part for TransferIdle is
+// dropped, its chunks with it; one that took a part since is armed again for the rest of its bound; one already gone is
+// nothing to do. It answers whether it dropped the transfer.
+func (h *Holder) Expired(id string) (bool, error) {
+	t, found, err := h.Transfers.Get(id)
+	if err != nil {
+		return false, fmt.Errorf("transfer %s's idle bound ended, but it cannot be read: %w", id, err)
+	}
+	if !found {
+		return false, nil
+	}
+	idle := h.now().Sub(t.Last)
+	if idle >= TransferIdle {
+		if _, err := h.Drop(id); err != nil {
+			return false, fmt.Errorf("transfer %s had no part for %s and is not dropped: %w", id, TransferIdle, err)
+		}
+		return true, nil
+	}
+	if h.Idle == nil {
+		return false, nil
+	}
+	return false, h.Idle(id, max(TransferIdle-idle, time.Second))
+}
+
 func checkPublish(seat string, r PublishRequest) error {
 	switch {
 	case !slices.Contains(Seats, seat):

modules/distribution/module.json

@@ -83,9 +83,14 @@
     {
       "name": "transfers",
       "storage": "key-value",
+      "template": "state"
+    },
+    {
+      "name": "idle",
+      "storage": "key-value",
       "template": "state",
       "options": {
-        "ttl-seconds": 86400
+        "key-ttl": true
       }
     },
     {