mesh-sdk/go/parts/receive_test.go

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 !RefusedAtCap(strings.TrimPrefix(pkgLog[len(pkgLog)-1].outcome, "refused: ")) || RefusedAtCap(strings.TrimPrefix(pkgLog[len(pkgLog)-2].outcome, "refused: ")) {
			t.Errorf("RefusedAtCap tells the cap's refusal from the check's: %q, %q", pkgLog[len(pkgLog)-1].outcome, pkgLog[len(pkgLog)-2].outcome)
		}
		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)
		}
	})
}