diff --git a/go/packages_list_test.go b/go/packages_list_test.go index d0f3b04..9540ac8 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -25,7 +25,7 @@ import ( var meshSDKPackages = []string{ ".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", "composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership", - "objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions", "publish", + "objectlink", "outward", "overlay", "pernode", "process", "provider", "parts", "provider/harness", "provisions", "publish", "sdkversion", "seal", "seatdef", "testbus", "views", } diff --git a/go/parts/receive.go b/go/parts/receive.go new file mode 100644 index 0000000..d2ac237 --- /dev/null +++ b/go/parts/receive.go @@ -0,0 +1,375 @@ +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 +// 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. + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "slices" + "sort" + "strings" + "time" +) + +// The bounds design 51 gives the receiving side; a Receiver's own fields override them. +const ( + // MaxReceived is the most bytes one chunk may carry when it arrives. + MaxReceived = 1 << 20 + // IdleAfter is how long a transfer with no chunk stays open. + IdleAfter = 24 * time.Hour + // FullPercent is the fill of the objects past which the receiver says they fill. + FullPercent = 80 +) + +// Want is one part an opening names: its digest (sha256:) and its size. +type Want struct { + Digest string `json:"digest"` + Size int64 `json:"size"` +} + +// Transfer is one open transfer, kept in the caller's state; its chunks are objects of the caller's object store. Shape +// is the caller's own body of the opening, kept as it came. +type Transfer[M any] struct { + ID string `json:"id"` + Opener string `json:"opener"` + Parts []Want `json:"parts"` + // Held are the parts whose every chunk arrived and hashed right, or that the holder held at the open; Chunks the + // chunks taken per part. + Held []string `json:"held,omitempty"` + Chunks map[string]int `json:"chunks,omitempty"` + Opened time.Time `json:"opened"` + Last time.Time `json:"last"` + Shape M `json:"shape"` +} + +// Lacks is the digests of the parts the transfer does not hold yet, sorted. +func (t Transfer[M]) Lacks() []string { + var out []string + for _, p := range t.Parts { + if !slices.Contains(t.Held, p.Digest) && !slices.Contains(out, p.Digest) { + out = append(out, p.Digest) + } + } + sort.Strings(out) + return out +} + +// Taken says whether any chunk of the part was taken in this transfer (a part held at the open was not). +func (t Transfer[M]) Taken(digest string) bool { + _, ok := t.Chunks[digest] + return ok +} + +// Objects is the holder's own object store the chunks go into. +type Objects interface { + Put(name string, data []byte) error + Get(name string) ([]byte, error) + Delete(name string) error + // Used is the bytes the store holds. + Used() (uint64, error) +} + +// Transfers is the holder's own state of open transfers, keyed by Transfer.ID. +type Transfers[M any] interface { + Get(id string) (Transfer[M], bool, error) + Put(t Transfer[M]) error + Delete(id string) error + List() ([]Transfer[M], error) +} + +// Received is the receiver's answer to an open or a chunk: the parts still lacked, and whether the transfer ended +// (stored or refused). +type Received struct { + ID string + Lacks []string + Ended bool +} + +// Receiver takes transfers in parts into a holder's own objects. +type Receiver[M any] struct { + Objects Objects + 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 uint64 + // MaxChunk, Idle and Full override MaxReceived, IdleAfter and FullPercent when set. + MaxChunk int + Idle time.Duration + Full uint64 + // Holds says the holder already holds a part, which then counts as held at the open and is not sent. Optional. + Holds func(t Transfer[M], digest string) bool + // Check is the caller's check of a whole transfer before anything is stored; its error is the refusal's reason. + // Optional. + Check func(t Transfer[M]) error + // Store stores a whole, checked transfer: the parts taken in it by digest (not those held at the open). It says + // the outcome itself; its error is answered as it is. + Store func(t Transfer[M], parts map[string][]byte) error + // Refuse says a refusal by the receiver (the cap, the caller's check, a part it cannot read back). Its own error is + // joined to the refusal's. + 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 +} + +func (r *Receiver[M]) now() time.Time { + if r.Now != nil { + return r.Now().UTC() + } + return time.Now().UTC() +} + +func (r *Receiver[M]) maxChunk() int { + if r.MaxChunk > 0 { + return r.MaxChunk + } + return MaxReceived +} + +func (r *Receiver[M]) idle() time.Duration { + if r.Idle > 0 { + return r.Idle + } + return IdleAfter +} + +func (r *Receiver[M]) fullPct() uint64 { + if r.Full > 0 { + return r.Full + } + return FullPercent +} + +// Open opens a transfer for opener, or opens it again: a transfer already open answers what it lacks, so a cut transfer +// goes on from there, and only its opener may open it again. A new transfer that would pass the cap is refused; parts +// the holder holds count as held; when nothing is lacked the transfer ends at once. +func (r *Receiver[M]) Open(id, opener string, parts []Want, shape M) (Received, error) { + switch { + case strings.TrimSpace(id) == "" || strings.ContainsAny(id, "/. *>"): + return Received{}, fmt.Errorf("transfer %q is no transfer id", id) + case len(parts) == 0: + return Received{}, errors.New("no parts") + } + for _, p := range parts { + if !strings.HasPrefix(p.Digest, "sha256:") || len(p.Digest) != len("sha256:")+64 { + return Received{}, fmt.Errorf("part digest %q is no sha256 digest", p.Digest) + } + } + if t, found, err := r.Transfers.Get(id); err != nil { + return Received{}, err + } else if found { + 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 + } + 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() + if err != nil { + return Received{}, fmt.Errorf("the objects' fill cannot be read: %w", err) + } + var size uint64 + 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("the objects are at their cap of %d bytes (%d held, %d asked)", r.Cap, used, size)) + } + if r.Holds != nil { + for _, p := range parts { + if r.Holds(t, p.Digest) { + t.Held = append(t.Held, p.Digest) + } + } + } + if err := r.Transfers.Put(t); err != nil { + return Received{}, fmt.Errorf("transfer %s cannot be kept: %w", id, err) + } + r.fill() + if len(t.Lacks()) == 0 { + return Received{ID: id, Ended: true}, r.finish(t) + } + return Received{ID: id, Lacks: t.Lacks()}, nil +} + +// Take takes one chunk of a part from the transfer's opener: chunks of a part arrive in sequence from 0, the last +// marked; the whole part must hash to its digest, else its chunks are dropped and the part is lacked again. Belongs, +// when given, refuses a transfer the caller's own shape says is not this call's (another seat's, say). When the last +// lacked part arrives the transfer ends: the caller's check, then its store. +func (r *Receiver[M]) Take(id, opener, digest string, sequence int, last bool, data []byte, belongs func(Transfer[M]) error) (Received, error) { + t, found, err := r.Transfers.Get(id) + switch { + case err != nil: + return Received{}, err + case !found: + return Received{}, fmt.Errorf("transfer %s is not open (dropped after %s with no chunk, or never opened)", id, r.idle()) + case t.Opener != opener: + return Received{}, fmt.Errorf("refused for %s: transfer %s was opened by another caller", opener, id) + } + if belongs != nil { + if err := belongs(t); err != nil { + return Received{}, err + } + } + switch { + case len(data) > r.maxChunk(): + return Received{}, fmt.Errorf("a chunk of %d bytes is over the %d a chunk may carry", len(data), r.maxChunk()) + case !slices.ContainsFunc(t.Parts, func(p Want) bool { return p.Digest == digest }): + return Received{}, fmt.Errorf("transfer %s names no part %s", id, digest) + case slices.Contains(t.Held, digest): + return Received{ID: id, Lacks: t.Lacks()}, nil + case sequence != t.Chunks[digest]: + return Received{}, fmt.Errorf("part %s of transfer %s waits for chunk %d, not %d", digest, id, t.Chunks[digest], sequence) + } + if err := r.Objects.Put(r.chunkName(id, digest, sequence), data); err != nil { + return Received{}, fmt.Errorf("the objects did not take the chunk: %w", err) + } + t.Chunks[digest] = sequence + 1 + t.Last = r.now() + if last { + whole, err := r.part(t, digest) + if err != nil { + return Received{}, err + } + if got := "sha256:" + sum(whole); got != digest { + r.dropChunks(t, digest) + delete(t.Chunks, digest) + _ = r.Transfers.Put(t) + r.fill() + return Received{}, fmt.Errorf("refused: the bytes of part %s hash to %s; send the part again", digest, got) + } + t.Held = append(t.Held, digest) + } + if err := r.Transfers.Put(t); err != nil { + return Received{}, err + } + r.fill() + if len(t.Lacks()) == 0 { + return Received{ID: id, Ended: true}, r.finish(t) + } + return Received{ID: id, Lacks: t.Lacks()}, nil +} + +// finish checks and stores a whole transfer; its chunks and its key go either way. +func (r *Receiver[M]) finish(t Transfer[M]) error { + defer func() { + for _, p := range t.Parts { + r.dropChunks(t, p.Digest) + } + _ = r.Transfers.Delete(t.ID) + r.fill() + }() + if r.Check != nil { + if err := r.Check(t); err != nil { + return r.refuse(t, err.Error()) + } + } + parts := map[string][]byte{} + for _, p := range t.Parts { + if !t.Taken(p.Digest) { + continue // held by the holder already + } + b, err := r.part(t, p.Digest) + if err != nil { + return r.refuse(t, "part "+p.Digest+" cannot be read from the objects: "+err.Error()) + } + parts[p.Digest] = b + } + return r.Store(t, parts) +} + +func (r *Receiver[M]) refuse(t Transfer[M], reason string) error { + err := errors.New("refused: " + reason) + if r.Refuse != nil { + if serr := r.Refuse(t, reason); serr != nil { + return fmt.Errorf("%w; and the refusal could not be said: %w", err, serr) + } + } + return err +} + +func (r *Receiver[M]) part(t Transfer[M], digest string) ([]byte, error) { + var buf bytes.Buffer + for seq := 0; seq < t.Chunks[digest]; seq++ { + b, err := r.Objects.Get(r.chunkName(t.ID, digest, seq)) + if err != nil { + return nil, err + } + buf.Write(b) + } + return buf.Bytes(), nil +} + +func (r *Receiver[M]) dropChunks(t Transfer[M], digest string) { + for seq := 0; seq < t.Chunks[digest]; seq++ { + _ = r.Objects.Delete(r.chunkName(t.ID, digest, seq)) + } +} + +func (r *Receiver[M]) chunkName(id, digest string, seq int) string { + return fmt.Sprintf("%s%s/%s/%d", r.Prefix, id, strings.TrimPrefix(digest, "sha256:"), seq) +} + +// fill tells Filled the objects' fill. +func (r *Receiver[M]) fill() { + if r.Filled == nil || r.Cap == 0 { + return + } + used, err := r.Objects.Used() + if err != nil { + return + } + r.Filled(used, r.Cap, used*100 >= r.Cap*r.fullPct()) +} + +// Drop drops a transfer, by hand or because it was idle: its chunks and its key. +func (r *Receiver[M]) Drop(id string) (Transfer[M], error) { + t, found, err := r.Transfers.Get(id) + if err != nil { + return Transfer[M]{}, err + } + if !found { + return Transfer[M]{}, fmt.Errorf("transfer %s is not open", id) + } + for _, p := range t.Parts { + r.dropChunks(t, p.Digest) + } + if err := r.Transfers.Delete(id); err != nil { + return Transfer[M]{}, err + } + r.fill() + 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. +func (r *Receiver[M]) DropIdle() ([]string, error) { + all, err := r.Transfers.List() + if err != nil { + return nil, err + } + var out []string + for _, t := range all { + if r.now().Sub(t.Last) >= r.idle() { + if _, err := r.Drop(t.ID); err == nil { + out = append(out, t.ID) + } + } + } + sort.Strings(out) + return out, nil +} + +func sum(b []byte) string { + s := sha256.Sum256(b) + return hex.EncodeToString(s[:]) +} diff --git a/go/parts/receive_test.go b/go/parts/receive_test.go new file mode 100644 index 0000000..0f355bf --- /dev/null +++ b/go/parts/receive_test.go @@ -0,0 +1,220 @@ +package parts + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "reflect" + "strings" + "testing" + "time" +) + +type memObjects map[string][]byte + +func (m memObjects) Put(n string, d []byte) error { m[n] = append([]byte(nil), d...); return nil } +func (m memObjects) Get(n string) ([]byte, error) { + if d, ok := m[n]; ok { + return d, nil + } + return nil, errors.New("no object " + n) +} +func (m memObjects) Delete(n string) error { delete(m, n); return nil } +func (m memObjects) Used() (uint64, error) { + var u uint64 + for _, d := range m { + u += uint64(len(d)) + } + return u, nil +} + +type memTransfers[M any] map[string]Transfer[M] + +func (m memTransfers[M]) Get(id string) (Transfer[M], bool, error) { t, ok := m[id]; return t, ok, nil } +func (m memTransfers[M]) Put(t Transfer[M]) error { m[t.ID] = t; return nil } +func (m memTransfers[M]) Delete(id string) error { delete(m, id); return nil } +func (m memTransfers[M]) List() ([]Transfer[M], error) { + var out []Transfer[M] + for _, t := range m { + out = append(out, t) + } + return out, nil +} + +// Two shapes of an opening, as two different seats' holders would keep them. +type packageShape struct{ Name, Version string } +type imageShape struct { + Repository string + Layers int +} + +type ended struct { + id, outcome string + parts map[string][]byte +} + +func digestOf(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) } + +func receiverOf[M any](objects memObjects, prefix string, now *time.Time, log *[]ended, check func(Transfer[M]) error) *Receiver[M] { + return &Receiver[M]{Objects: objects, Transfers: memTransfers[M]{}, Prefix: prefix, Cap: 1 << 20, MaxChunk: 1000, + Check: check, Now: func() time.Time { return *now }, + Store: func(t Transfer[M], parts map[string][]byte) error { + *log = append(*log, ended{t.ID, "stored", parts}) + return nil + }, + Refuse: func(t Transfer[M], reason string) error { + *log = append(*log, ended{t.ID, "refused: " + reason, nil}) + return nil + }} +} + +// TestAHolderTakesAFileInParts runs the receiving side for two holders with two shapes and two prefixes on one object +// store (ADR 0627 §2): neither names a seat; the shapes, the checks and the saying are each caller's own. +func TestAHolderTakesAFileInParts(t *testing.T) { + objects := memObjects{} + now := time.Unix(1000, 0) + var pkgLog, imgLog []ended + pkg := receiverOf(objects, "npm/", &now, &pkgLog, func(t Transfer[packageShape]) error { + if t.Shape.Version == "" { + return errors.New("the builder did not number it") + } + return nil + }) + img := receiverOf[imageShape](objects, "image/", &now, &imgLog, nil) + blob := []byte(strings.Repeat("a", 1500)) + part := Want{Digest: digestOf(blob), Size: int64(len(blob))} + layer := []byte("layer") + held := Want{Digest: digestOf(layer), Size: int64(len(layer))} + img.Holds = func(_ Transfer[imageShape], d string) bool { return d == held.Digest } + + t.Run("two holders, one transfer id, two shapes", func(t *testing.T) { + a, err := pkg.Open("t1", "builder", []Want{part}, packageShape{"m/program", "0.1.4"}) + if err != nil || !reflect.DeepEqual(a.Lacks, []string{part.Digest}) { + t.Fatalf("pkg open = %+v, %v", a, err) + } + b, err := img.Open("t1", "builder", []Want{part, held}, imageShape{"m/app", 2}) + if err != nil || !reflect.DeepEqual(b.Lacks, []string{part.Digest}) { + t.Fatalf("img open = %+v, %v: a part the holder holds is not lacked", b, err) + } + for _, r := range []func(int, bool, []byte) (Received, error){ + func(s int, l bool, d []byte) (Received, error) { + return pkg.Take("t1", "builder", part.Digest, s, l, d, nil) + }, + func(s int, l bool, d []byte) (Received, error) { + return img.Take("t1", "builder", part.Digest, s, l, d, nil) + }, + } { + if _, err := r(0, false, blob[:1000]); err != nil { + t.Fatal(err) + } + } + // A cut transfer opened again goes on from what is held; another caller may not. + if a, _ := pkg.Open("t1", "builder", []Want{part}, packageShape{}); !reflect.DeepEqual(a.Lacks, []string{part.Digest}) { + t.Errorf("reopen = %+v", a) + } + if _, err := pkg.Open("t1", "other", []Want{part}, packageShape{}); err == nil || !strings.Contains(err.Error(), "another caller") { + t.Errorf("reopen by another = %v", err) + } + if _, err := pkg.Take("t1", "builder", part.Digest, 0, false, blob[:1000], nil); err == nil || !strings.Contains(err.Error(), "waits for chunk 1") { + t.Errorf("a chunk out of sequence = %v", err) + } + a, err = pkg.Take("t1", "builder", part.Digest, 1, true, blob[1000:], nil) + if err != nil || !a.Ended || len(pkgLog) != 1 || pkgLog[0].outcome != "stored" || string(pkgLog[0].parts[part.Digest]) != string(blob) { + t.Fatalf("pkg end = %+v, %v, %+v", a, err, pkgLog) + } + // The image holder's chunks stood apart under its own prefix and are still there. + if _, ok := objects["image/t1/"+strings.TrimPrefix(part.Digest, "sha256:")+"/0"]; !ok || len(objects) != 1 { + t.Fatalf("objects after pkg's end = %v keys, want only the image holder's chunk", len(objects)) + } + b, err = img.Take("t1", "builder", part.Digest, 1, true, blob[1000:], func(t Transfer[imageShape]) error { return nil }) + if err != nil || !b.Ended || len(imgLog) != 1 || len(imgLog[0].parts) != 1 { + t.Fatalf("img end = %+v, %v, %+v: only the part taken is stored", b, err, imgLog) + } + if len(objects) != 0 { + t.Errorf("objects after both ends = %d, want none", len(objects)) + } + }) + + t.Run("refusals", func(t *testing.T) { + for _, c := range []struct { + name, want string + do func() error + }{ + {"no transfer id", "no transfer id", func() error { _, err := pkg.Open("a/b", "builder", []Want{part}, packageShape{}); return err }}, + {"no sha256 digest", "no sha256 digest", func() error { + _, err := pkg.Open("t", "builder", []Want{{Digest: "md5:x"}}, packageShape{}) + return err + }}, + {"another caller's chunk", "another caller", func() error { + _, _ = pkg.Open("t2", "builder", []Want{part}, packageShape{"m", "1"}) + _, err := pkg.Take("t2", "other", part.Digest, 0, true, blob, nil) + return err + }}, + {"a chunk over the bound", "over the 1000", func() error { + _, err := pkg.Take("t2", "builder", part.Digest, 0, false, make([]byte, 1001), nil) + return err + }}, + {"a part the transfer does not name", "names no part", func() error { + _, err := pkg.Take("t2", "builder", held.Digest, 0, true, layer, nil) + return err + }}, + {"the caller's own belongs", "not this seat's", func() error { + _, err := pkg.Take("t2", "builder", part.Digest, 0, true, blob, func(Transfer[packageShape]) error { return errors.New("not this seat's") }) + return err + }}, + {"bytes that do not hash", "send the part again", func() error { + _, err := pkg.Take("t2", "builder", part.Digest, 0, true, []byte("other"), nil) + return err + }}, + {"the caller's check", "refused: the builder did not number it", func() error { + _, _ = pkg.Open("t3", "builder", []Want{part}, packageShape{Name: "m"}) + _, _ = pkg.Take("t3", "builder", part.Digest, 0, false, blob[:1000], nil) + _, err := pkg.Take("t3", "builder", part.Digest, 1, true, blob[1000:], nil) + return err + }}, + {"the cap", "at their cap", func() error { + pkg.Cap = 100 + defer func() { pkg.Cap = 1 << 20 }() + _, err := pkg.Open("t4", "builder", []Want{part}, packageShape{}) + return err + }}, + } { + if err := c.do(); err == nil || !strings.Contains(err.Error(), c.want) { + t.Errorf("%s: err = %v, want %q", c.name, err, c.want) + } + } + if got := pkgLog[len(pkgLog)-1].outcome; !strings.HasPrefix(got, "refused: the objects are at their cap") { + t.Errorf("the cap's refusal said %q", got) + } + if got := pkgLog[len(pkgLog)-2].outcome; got != "refused: the builder did not number it" { + t.Errorf("the check's refusal said %q", got) + } + // A part that did not hash is lacked again, its chunk gone. + if a, _ := pkg.Open("t2", "builder", []Want{part}, packageShape{}); !reflect.DeepEqual(a.Lacks, []string{part.Digest}) { + t.Errorf("after a bad hash = %+v", a) + } + }) + + t.Run("idle drop and the fill", func(t *testing.T) { + var fills []bool + pkg.Filled = func(used, cap uint64, full bool) { fills = append(fills, full) } + pkg.Cap, pkg.Full = 1500, 60 + defer func() { pkg.Cap, pkg.Full, pkg.Filled = 1<<20, 0, nil }() + _, _ = pkg.Open("t5", "builder", []Want{part}, packageShape{"m", "1"}) + _, _ = pkg.Take("t5", "builder", part.Digest, 0, false, blob[:1000], nil) + if len(fills) == 0 || !fills[len(fills)-1] { + t.Errorf("fills = %v, want full at 1000 of 1500 bytes (60 percent)", fills) + } + now = now.Add(IdleAfter) + dropped, err := pkg.DropIdle() + if err != nil || !reflect.DeepEqual(dropped, []string{"t2", "t5"}) { + t.Errorf("DropIdle = %v, %v", dropped, err) + } + if fills[len(fills)-1] || len(objects) != 0 { + t.Errorf("after the drop: full %v, %d objects", fills[len(fills)-1], len(objects)) + } + if _, err := pkg.Take("t5", "builder", part.Digest, 1, true, blob[1000:], nil); err == nil || !strings.Contains(err.Error(), "not open") { + t.Errorf("a chunk after the drop = %v", err) + } + }) +}