novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

F-parts: distribution and gitea bind mesh-sdk's idle key and binding cap instead of their own copies

Browse Source 
Both holders carried the same idle bound and the same read of incoming's cap; the second instance of the
kind moved into mesh-sdk's Receiver (Arm, Expired, Bound, with BusArm, BusCap and FollowIdle). Each
holder now hands its idle state and its objects to those, and keeps its rows.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 17:26:11 +01:00
parent d320a19a8c
commit f28d3bf286
5 changed files with 37 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/distribution/cmd/distribution-tools/main.go
+8 -51
View File
@@ -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)
modules/distribution/internal/holder/transfer.go
+10 -44
View File
@@ -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 {