diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index dd32a367..17b9ec2c 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/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) diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index c5efea76..0b384a03 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/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]) + } }) } diff --git a/modules/distribution/internal/holder/transfer.go b/modules/distribution/internal/holder/transfer.go index 6df69f08..8882cf01 100644 --- a/modules/distribution/internal/holder/transfer.go +++ b/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): diff --git a/modules/distribution/module.json b/modules/distribution/module.json index f3e98583..8f7e20dc 100644 --- a/modules/distribution/module.json +++ b/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 } }, {