diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 98603529..de40893b 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -15,12 +15,10 @@ 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" @@ -166,53 +164,6 @@ 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{} @@ -547,10 +498,16 @@ 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), Cap: incomingCap(), Idle: armIdle} + Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState), Cap: parts.BusCap(holder.IncomingObjects), Idle: parts.BusArm(holder.IdleState)} 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) + go parts.FollowIdle(holder.IdleState, h.Expired, func(id string, dropped bool, err error) { + if err != nil { + logf("%v", err) + return + } + logf("transfer %s had no part for %s: dropped with its parts", id, holder.TransferIdle) + }) 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/transfer.go b/modules/distribution/internal/holder/transfer.go index 8882cf01..599b105f 100644 --- a/modules/distribution/internal/holder/transfer.go +++ b/modules/distribution/internal/holder/transfer.go @@ -162,9 +162,10 @@ type Holder struct { Say Say Finding Finding Now func() time.Time - // Cap reads incoming's cap from its binding; IncomingCap when nil or when the binding names none. + // Cap reads incoming's cap from its binding (parts.BusCap); 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 writes a transfer's idle key with its own time to live, once (parts.BusArm over IdleState); the receiver + // arms it on each open that leaves a transfer open. Nil arms nothing. Idle func(id string, ttl time.Duration) error } @@ -176,22 +177,13 @@ func (h *Holder) now() time.Time { } // 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() } +func (h *Holder) CapNow() uint64 { return h.receiver().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] { - return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: h.cap(), MaxChunk: MaxPart, - Idle: TransferIdle, Full: IncomingFullPct, Now: h.now, + return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: IncomingCap, Bound: h.Cap, + MaxChunk: MaxPart, Idle: TransferIdle, Arm: h.Idle, Full: IncomingFullPct, Now: h.now, // Only the missing layers are sent: a part the store already holds under the name counts as held. Holds: func(t Transfer, digest string) bool { ok, err := h.Store.Has(t.Shape.Name, digest) @@ -268,13 +260,6 @@ 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 } @@ -307,29 +292,10 @@ func (h *Holder) Drop(id string) (Transfer, error) { return h.receiver().Drop(id // 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)) -} +// Expired is told that a transfer's idle key ended (the bus's marker, parts.FollowIdle): the receiver drops a transfer +// with no part for TransferIdle, its chunks with it, arms one that took a part since again for the rest, and leaves one +// already gone. It answers whether it dropped the transfer. +func (h *Holder) Expired(id string) (bool, error) { return h.receiver().Expired(id) } func checkPublish(seat string, r PublishRequest) error { switch { diff --git a/modules/gitea/cmd/package-registry/incoming.go b/modules/gitea/cmd/package-registry/incoming.go index eda80b2a..9ad5c3c7 100644 --- a/modules/gitea/cmd/package-registry/incoming.go +++ b/modules/gitea/cmd/package-registry/incoming.go @@ -10,11 +10,9 @@ import ( "fmt" "os" "strings" - "sync/atomic" - "time" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/busstate" + "git.novox.be/novox/mesh-sdk/go/parts" "gitea/internal/packageseats" ) @@ -53,53 +51,19 @@ func incomingTool(typ, seat string, holder func() (*packageseats.Holder, error)) } } -// incomingCap reads incoming's cap from its binding (the bus's bound of the object store), once it answers one. -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(packageseats.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(key string, ttl time.Duration) error { - if _, err := stdio.State(packageseats.IdleState).CreateTTL(key, map[string]any{"transfer": key}, ttl); err != nil && !errors.Is(err, stdio.ErrChanged) { - return err - } - return nil -} - // followIdle drops a transfer when its idle key ends (packageseats.Holder.Expired): the bus's marker is the event. func followIdle(holder func() (*packageseats.Holder, error)) { - <-stdio.Served() - stdio.State(packageseats.IdleState).Follow("", func(ch stdio.StateChange) error { - if ch.Op != stdio.OpExpire { - return nil - } + parts.FollowIdle(packageseats.IdleState, func(id string) (bool, error) { h, err := holder() if err != nil { - fmt.Fprintf(os.Stderr, "transfer %s's idle bound ended, but the holder cannot be made: %v\n", ch.Key, err) - return err + return false, fmt.Errorf("transfer %s's idle bound ended, but the holder cannot be made: %w", id, err) } - dropped, err := h.Expired(ch.Key) + return h.Expired(id) + }, func(id string, dropped bool, err error) { if err != nil { fmt.Fprintln(os.Stderr, err) - return err + return } - if dropped { - fmt.Fprintf(os.Stderr, "transfer %s had no chunk for %s: dropped with its chunks\n", ch.Key, packageseats.TransferIdle) - } - return nil + fmt.Fprintf(os.Stderr, "transfer %s had no chunk for %s: dropped with its chunks\n", id, packageseats.TransferIdle) }) } diff --git a/modules/gitea/cmd/package-registry/seats.go b/modules/gitea/cmd/package-registry/seats.go index 6447791f..36a0ff90 100644 --- a/modules/gitea/cmd/package-registry/seats.go +++ b/modules/gitea/cmd/package-registry/seats.go @@ -264,7 +264,7 @@ func finding(kind, subject, words string, raised bool) { } // boundCap is incoming's cap from its binding, read once per bundle. -var boundCap = incomingCap() +var boundCap = parts.BusCap(packageseats.IncomingObjects) // busHolder is the holder over the runtime's bus state, the builder's numbering and the forge. func busHolder(forge func() (*Forge, error)) (*packageseats.Holder, error) { @@ -275,7 +275,7 @@ func busHolder(forge func() (*Forge, error)) (*packageseats.Holder, error) { return &packageseats.Holder{Callers: callers, Objects: parts.BusObjects(packageseats.IncomingObjects), Transfers: parts.BusTransfers[packageseats.Opening](packageseats.TransfersState), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: seatStore{forge}, Say: say, Finding: finding, - Cap: boundCap, Idle: armIdle}, nil + Cap: boundCap, Idle: parts.BusArm(packageseats.IdleState)}, nil } // SeatTools are both seats' verbs as tools .. diff --git a/modules/gitea/internal/packageseats/holder.go b/modules/gitea/internal/packageseats/holder.go index 80ae121e..557fa181 100644 --- a/modules/gitea/internal/packageseats/holder.go +++ b/modules/gitea/internal/packageseats/holder.go @@ -154,21 +154,15 @@ type Holder struct { // Finding writes (raised) or clears the module's finding of a kind for a subject. Finding func(kind, subject, words string, raised bool) Now func() time.Time - // Cap reads incoming's cap from its binding; IncomingCap when nil or when the binding names none. + // Cap reads incoming's cap from its binding (parts.BusCap); 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 writes a transfer's idle key with its own time to live, once (parts.BusArm over IdleState); the receiver + // arms it on each open that leaves a transfer open. Nil arms nothing. Idle func(key string, ttl time.Duration) error } // 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) CapNow() uint64 { return h.receiver().CapNow() } func (h *Holder) now() time.Time { if h.Now != nil { @@ -184,9 +178,8 @@ func Reference(name, version string) string { return name + "@" + version } func key(seat, transfer string) string { return seat + "_" + transfer } func (h *Holder) receiver() *parts.Receiver[Opening] { - cap := h.CapNow() - return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: cap, MaxChunk: MaxChunk, - Idle: TransferIdle, Full: IncomingFullPct, Now: h.now, + return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: IncomingCap, Bound: h.Cap, + MaxChunk: MaxChunk, Idle: TransferIdle, Arm: h.Idle, Full: IncomingFullPct, Now: h.now, // The builder's numbering is asked before anything is uploaded (ADR 0605 ยง4). Check: func(t Transfer) error { ok, why, err := h.Builder.Recorded(t.Shape.Name, t.Shape.Version, t.Shape.Digest) @@ -265,13 +258,6 @@ func (h *Holder) Publish(seat, caller string, p Publish) (Answer, error) { if err != nil { return h.answer(p.Transfer, got), fmt.Errorf("publish: %w", err) } - // An open transfer is bounded by its idle key; one that cannot be written fails the publish loudly, and the build - // module's next publish of the same transfer answers the part held and arms it again. - if !got.Ended && h.Idle != nil { - if err := h.Idle(key(seat, p.Transfer), TransferIdle); err != nil { - return h.answer(p.Transfer, got), fmt.Errorf("publish: transfer %s is open but its idle bound is not kept, so it would never be dropped: %w", p.Transfer, err) - } - } return h.answer(p.Transfer, got), nil } @@ -378,29 +364,10 @@ func (h *Holder) IncomingOf(seat string) (Incoming, error) { return out, nil } -// Expired is told that a transfer's idle key (its key in the state transfers) ended: a transfer with no chunk for -// TransferIdle is dropped with its chunks; one that took a chunk since is armed again for the rest; one gone is -// nothing to do. It answers whether it dropped it. -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.receiver().Drop(id); err != nil { - return false, fmt.Errorf("transfer %s had no chunk 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)) -} +// Expired is told that a transfer's idle key (its key in the state transfers) ended (the bus's marker, +// parts.FollowIdle): the receiver drops a transfer with no chunk for TransferIdle with its chunks, arms one that took a +// chunk since again for the rest, and leaves one gone. It answers whether it dropped it. +func (h *Holder) Expired(id string) (bool, error) { return h.receiver().Expired(id) } func checkPublish(seat string, p Publish) error { switch {