R1-F-parts: the receiving side keeps a transfer's idle key and reads its cap from the binding
distribution and gitea each carried the same idle bound (one expiring key per open transfer, its end dropping a transfer idle for 24 hours or arming it again for the rest) and the same read of incoming's cap from its binding: the second instance of the kind, so it moves to the Receiver every holder shares. Arm is called on each open that leaves a transfer open, a key not written failing it loudly; Expired handles a key's end; Bound reads the binding's cap over Cap. BusArm, BusCap and FollowIdle bind them to the holder's own state and objects. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -3,6 +3,9 @@ package parts
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/busstate"
|
||||
@@ -63,3 +66,53 @@ func (b busTransfers[M]) List() ([]Transfer[M], error) {
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// BusCap is a Receiver's Bound over the holder's own object store of that name: the cap its binding carries (the bus's
|
||||
// own bound, the store's max bytes), read through the runtime once it answers one, since a binding's cap does not
|
||||
// change while the bundle runs. Make it once per bundle, not per receiver.
|
||||
func BusCap(local string) func() (uint64, error) {
|
||||
var known atomic.Uint64
|
||||
return func() (uint64, error) {
|
||||
if c := known.Load(); c > 0 {
|
||||
return c, nil
|
||||
}
|
||||
f, err := busstate.ObjectsOf(local).Fill()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s's cap cannot be read from its binding: %w", local, err)
|
||||
}
|
||||
if f.MaxBytes == 0 {
|
||||
return 0, fmt.Errorf("%s's binding carries no cap", local)
|
||||
}
|
||||
known.Store(f.MaxBytes)
|
||||
return f.MaxBytes, nil
|
||||
}
|
||||
}
|
||||
|
||||
// BusArm is a Receiver's Arm over the holder's own key-value state of that name, declared key-ttl: one key per open
|
||||
// transfer with its own time to live; a key already there is the same bound, kept.
|
||||
func BusArm(state string) func(id string, ttl time.Duration) error {
|
||||
return func(id string, ttl time.Duration) error {
|
||||
if _, err := stdio.State(state).CreateTTL(id, map[string]any{"transfer": id}, ttl); err != nil && !errors.Is(err, stdio.ErrChanged) {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// FollowIdle follows the holder's idle state (the one BusArm writes) once the bundle serves, and hands the end of each
|
||||
// key (the bus's marker, OpExpire) to expired, a Receiver's Expired: the event drops an idle transfer, no timer runs.
|
||||
// said is told each outcome (dropped, or the error that stopped it) for the holder's log. It blocks; run it in its own
|
||||
// goroutine.
|
||||
func FollowIdle(state string, expired func(id string) (bool, error), said func(id string, dropped bool, err error)) {
|
||||
<-stdio.Served()
|
||||
stdio.State(state).Follow("", func(ch stdio.StateChange) error {
|
||||
if ch.Op != stdio.OpExpire {
|
||||
return nil
|
||||
}
|
||||
dropped, err := expired(ch.Key)
|
||||
if said != nil && (dropped || err != nil) {
|
||||
said(ch.Key, dropped, err)
|
||||
}
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
+66
-11
@@ -2,7 +2,7 @@ package parts
|
||||
|
||||
// The receiving side of a file in parts (novox/hq ADR 0611, ADR 0627 §2, design 51 *Publishing*): a holder opens a
|
||||
// transfer naming every part, takes the chunks of the parts it lacks into its own objects, checks each whole part's
|
||||
// hash, drops a transfer left idle and says when its objects fill. Like the sender it names no seat, no verb and no
|
||||
// hash, drops a transfer left idle when its idle bound ends and says when its objects fill. Like the sender it names no seat, no verb and no
|
||||
// seat's type: the caller's own shape of an opening rides along as M, its objects and its state are its own, and the
|
||||
// checks before storing, the storing and the saying of a refusal are the caller's functions. Each holder binds it to
|
||||
// the seat it holds.
|
||||
@@ -106,8 +106,12 @@ type Receiver[M any] struct {
|
||||
Transfers Transfers[M]
|
||||
// Prefix starts the name of every chunk object, so two receivers can share one object store.
|
||||
Prefix string
|
||||
// Cap is the bytes the objects may hold; an opening that would pass it is refused. Required.
|
||||
// Cap is the bytes the objects may hold; an opening that would pass it is refused. Required: it is said where
|
||||
// Bound answers no cap.
|
||||
Cap uint64
|
||||
// Bound reads the cap the objects' binding carries (BusCap): the bus's own bound, which then counts over Cap.
|
||||
// Optional; an error or a zero falls back to Cap.
|
||||
Bound func() (uint64, error)
|
||||
// MaxChunk, Idle and Full override MaxReceived, IdleAfter and FullPercent when set.
|
||||
MaxChunk int
|
||||
Idle time.Duration
|
||||
@@ -125,7 +129,11 @@ type Receiver[M any] struct {
|
||||
Refuse func(t Transfer[M], reason string) error
|
||||
// Filled is told the objects' fill after every change: used, the cap and whether it is past Full percent. Optional.
|
||||
Filled func(used, cap uint64, full bool)
|
||||
Now func() time.Time
|
||||
// Arm writes a transfer's idle key with its own time to live, once (BusArm over a state declared key-ttl); one
|
||||
// already there is no fault. The key's end is the event that calls Expired, so an open transfer left with no
|
||||
// chunk is dropped without a timer. Optional: nil arms nothing, and only DropIdle drops.
|
||||
Arm func(id string, ttl time.Duration) error
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func (r *Receiver[M]) now() time.Time {
|
||||
@@ -149,6 +157,16 @@ func (r *Receiver[M]) idle() time.Duration {
|
||||
return IdleAfter
|
||||
}
|
||||
|
||||
// CapNow is the cap the receiver counts against: the binding's (Bound) when it answers one, else Cap.
|
||||
func (r *Receiver[M]) CapNow() uint64 {
|
||||
if r.Bound != nil {
|
||||
if c, err := r.Bound(); err == nil && c > 0 {
|
||||
return c
|
||||
}
|
||||
}
|
||||
return r.Cap
|
||||
}
|
||||
|
||||
func (r *Receiver[M]) fullPct() uint64 {
|
||||
if r.Full > 0 {
|
||||
return r.Full
|
||||
@@ -177,7 +195,7 @@ func (r *Receiver[M]) Open(id, opener string, parts []Want, shape M) (Received,
|
||||
if t.Opener != opener {
|
||||
return Received{}, fmt.Errorf("refused for %s: transfer %s was opened by another caller", opener, id)
|
||||
}
|
||||
return Received{ID: id, Lacks: t.Lacks()}, nil
|
||||
return Received{ID: id, Lacks: t.Lacks()}, r.arm(id)
|
||||
}
|
||||
t := Transfer[M]{ID: id, Opener: opener, Parts: parts, Chunks: map[string]int{}, Opened: r.now(), Last: r.now(), Shape: shape}
|
||||
used, err := r.Objects.Used()
|
||||
@@ -188,8 +206,8 @@ func (r *Receiver[M]) Open(id, opener string, parts []Want, shape M) (Received,
|
||||
for _, p := range parts {
|
||||
size += uint64(max(p.Size, 0))
|
||||
}
|
||||
if used+size > r.Cap {
|
||||
return Received{ID: id, Ended: true}, r.refuse(t, fmt.Sprintf(AtCap+" of %d bytes (%d held, %d asked)", r.Cap, used, size))
|
||||
if cap := r.CapNow(); used+size > cap {
|
||||
return Received{ID: id, Ended: true}, r.refuse(t, fmt.Sprintf(AtCap+" of %d bytes (%d held, %d asked)", cap, used, size))
|
||||
}
|
||||
if r.Holds != nil {
|
||||
for _, p := range parts {
|
||||
@@ -205,7 +223,19 @@ func (r *Receiver[M]) Open(id, opener string, parts []Want, shape M) (Received,
|
||||
if len(t.Lacks()) == 0 {
|
||||
return Received{ID: id, Ended: true}, r.finish(t)
|
||||
}
|
||||
return Received{ID: id, Lacks: t.Lacks()}, nil
|
||||
return Received{ID: id, Lacks: t.Lacks()}, r.arm(id)
|
||||
}
|
||||
|
||||
// arm bounds an open transfer by its idle key. A key that cannot be written fails the open loudly (the transfer is
|
||||
// kept, so the opener's next open of it answers what is held and arms it again).
|
||||
func (r *Receiver[M]) arm(id string) error {
|
||||
if r.Arm == nil {
|
||||
return nil
|
||||
}
|
||||
if err := r.Arm(id, r.idle()); err != nil {
|
||||
return fmt.Errorf("transfer %s is open but its idle bound is not kept, so it would never be dropped: %w", id, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Take takes one chunk of a part from the transfer's opener: chunks of a part arrive in sequence from 0, the last
|
||||
@@ -332,14 +362,15 @@ func (r *Receiver[M]) chunkName(id, digest string, seq int) string {
|
||||
|
||||
// fill tells Filled the objects' fill.
|
||||
func (r *Receiver[M]) fill() {
|
||||
if r.Filled == nil || r.Cap == 0 {
|
||||
cap := r.CapNow()
|
||||
if r.Filled == nil || cap == 0 {
|
||||
return
|
||||
}
|
||||
used, err := r.Objects.Used()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
r.Filled(used, r.Cap, used*100 >= r.Cap*r.fullPct())
|
||||
r.Filled(used, cap, used*100 >= cap*r.fullPct())
|
||||
}
|
||||
|
||||
// Drop drops a transfer, by hand or because it was idle: its chunks and its key.
|
||||
@@ -361,8 +392,32 @@ func (r *Receiver[M]) Drop(id string) (Transfer[M], error) {
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// DropIdle drops every transfer with no chunk for Idle and answers their ids. It runs when its caller calls it (on an
|
||||
// open, on a listing), never on a timer.
|
||||
// Expired is told that a transfer's idle key ended (the bus's marker, FollowIdle): a transfer with no chunk for Idle
|
||||
// is dropped with its chunks; one that took a chunk 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 (r *Receiver[M]) Expired(id string) (bool, error) {
|
||||
t, found, err := r.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 := r.now().Sub(t.Last)
|
||||
if idle >= r.idle() {
|
||||
if _, err := r.Drop(id); err != nil {
|
||||
return false, fmt.Errorf("transfer %s had no chunk for %s and is not dropped: %w", id, r.idle(), err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
if r.Arm == nil {
|
||||
return false, nil
|
||||
}
|
||||
return false, r.Arm(id, max(r.idle()-idle, time.Second))
|
||||
}
|
||||
|
||||
// DropIdle drops every transfer with no chunk for Idle and answers their ids: a look by hand (a listing tool). The
|
||||
// drop itself follows each transfer's idle key (Arm, Expired), never a timer.
|
||||
func (r *Receiver[M]) DropIdle() ([]string, error) {
|
||||
all, err := r.Transfers.List()
|
||||
if err != nil {
|
||||
|
||||
@@ -238,4 +238,40 @@ func aHolderTakesAFileInParts(t *testing.T) {
|
||||
t.Errorf("a chunk after the drop = %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
// An open transfer is bounded by its idle key (Arm); the key's end (Expired) drops it idle, arms it again for the
|
||||
// rest when it took a chunk since, and leaves a gone one; a key not written fails the open loudly. The cap the
|
||||
// binding carries (Bound) counts over Cap.
|
||||
t.Run("the idle key and the binding's cap", func(t *testing.T) {
|
||||
armed := map[string]time.Duration{}
|
||||
pkg.Arm = func(id string, ttl time.Duration) error { armed[id] = ttl; return nil }
|
||||
defer func() { pkg.Arm, pkg.Bound = nil, nil }()
|
||||
if _, err := pkg.Open("t6", "builder", []Want{part}, packageShape{"m", "1"}); err != nil || armed["t6"] != IdleAfter {
|
||||
t.Fatalf("open = %v, armed %v; want t6's idle key for %s", err, armed, IdleAfter)
|
||||
}
|
||||
now = now.Add(time.Hour)
|
||||
_, _ = pkg.Take("t6", "builder", part.Digest, 0, false, blob[:1000], nil)
|
||||
now = now.Add(IdleAfter - time.Hour)
|
||||
if dropped, err := pkg.Expired("t6"); err != nil || dropped || armed["t6"] != time.Hour {
|
||||
t.Errorf("expired after a chunk = %v, %v, armed %v; want armed again for the hour left", dropped, err, armed["t6"])
|
||||
}
|
||||
now = now.Add(time.Hour)
|
||||
if dropped, err := pkg.Expired("t6"); err != nil || !dropped || len(objects) != 0 {
|
||||
t.Errorf("expired idle = %v, %v, %d objects; want dropped with its chunks", dropped, err, len(objects))
|
||||
}
|
||||
if dropped, err := pkg.Expired("t6"); err != nil || dropped {
|
||||
t.Errorf("expired once gone = %v, %v; want nothing to do", dropped, err)
|
||||
}
|
||||
pkg.Arm = func(string, time.Duration) error { return errors.New("bus down") }
|
||||
if _, err := pkg.Open("t7", "builder", []Want{part}, packageShape{"m", "1"}); err == nil || !strings.Contains(err.Error(), "never be dropped") {
|
||||
t.Errorf("open with no idle key = %v, want a loud failure", err)
|
||||
}
|
||||
pkg.Bound = func() (uint64, error) { return 100, nil }
|
||||
if _, err := pkg.Open("t8", "builder", []Want{part}, packageShape{}); err == nil || !strings.Contains(err.Error(), "cap of 100 bytes") {
|
||||
t.Errorf("open past the binding's cap = %v, want refused at 100 bytes", err)
|
||||
}
|
||||
if pkg.Bound = func() (uint64, error) { return 0, errors.New("no binding") }; pkg.CapNow() != pkg.Cap {
|
||||
t.Errorf("CapNow with no binding read = %d, want Cap %d", pkg.CapNow(), pkg.Cap)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user