221 lines
9.7 KiB
Go
221 lines
9.7 KiB
Go
package versions
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
|
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
|
)
|
|
|
|
type queueMem struct {
|
|
*reserveMem
|
|
bytes uint64
|
|
}
|
|
|
|
func (q *queueMem) Bytes(context.Context) (uint64, error) { return q.bytes, nil }
|
|
|
|
// republishBed is a Republisher over memory states: reservations of module m (a bundle by the oci seat and a Go package by
|
|
// the Go seat), one holder module `oci` that claims the oci seat, and a Send and a Unreleased that record.
|
|
type republishBed struct {
|
|
*Republisher
|
|
reserved *reserveMem
|
|
queue *queueMem
|
|
acted *reserveMem
|
|
now time.Time
|
|
sent []Publish
|
|
said []string
|
|
refuse string // the words Send fails with, empty for none
|
|
}
|
|
|
|
func newRepublishBed(t *testing.T, max int) *republishBed {
|
|
t.Helper()
|
|
b := &republishBed{reserved: &reserveMem{mem: newMem()}, queue: &queueMem{reserveMem: &reserveMem{mem: newMem()}}, acted: &reserveMem{mem: newMem()}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)}
|
|
res := &Reserver{Store: b.reserved, Now: func() time.Time { return b.now }, ReleasedOf: func(context.Context, string) ([]string, error) { return nil, nil }}
|
|
b.Republisher = &Republisher{Reserved: res, Queue: b.queue, Acted: b.acted, Max: max, Now: func() time.Time { return b.now }, Go: func(f func()) { f() },
|
|
Entry: func(_ context.Context, fp string) (meshcatalog.CatalogEntry, error) {
|
|
if fp == "src2:holder" {
|
|
return meshcatalog.CatalogEntry{Serves: []string{SeatArtifacts + ".publish", SeatArtifacts + ".versions"}}, nil
|
|
}
|
|
m := strings.TrimPrefix(fp, "src2:")
|
|
return meshcatalog.CatalogEntry{Artifacts: []meshcatalog.Artifact{
|
|
{Ref: m + ":prog", Kind: "bundle", Language: "go"},
|
|
{Ref: m + ":lib", Kind: "package", Language: "go", Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: "git.novox.be/novox/" + m + "/go"}}}}, nil
|
|
},
|
|
Send: func(_ context.Context, p Publish) error {
|
|
b.sent = append(b.sent, p)
|
|
if b.refuse != "" {
|
|
return errors.New(b.refuse)
|
|
}
|
|
return nil
|
|
},
|
|
Unreleased: func(_ context.Context, v ReservedVersion, why string) {
|
|
b.said = append(b.said, v.Module+" "+v.Version+": "+why)
|
|
},
|
|
Logf: t.Logf}
|
|
return b
|
|
}
|
|
|
|
// reserve reserves and builds a version of module m (its patch given by the reservation order), its build ended long ago.
|
|
func (b *republishBed) reserve(t *testing.T, module string) ReservedVersion {
|
|
t.Helper()
|
|
b.now = b.now.Add(time.Minute)
|
|
ctx := context.Background()
|
|
v, err := b.Reserved.Reserve(ctx, module, "0.1", []string{module + ":prog", module + ":lib"}, Source{Fingerprint: "src2:" + module})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for ref, d := range map[string]string{module + ":prog": "sha256:aa", module + ":lib": "sha256:bb"} {
|
|
if _, err := b.Reserved.Built(ctx, module, v.Version, ref, d, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
b.now = b.now.Add(UnreleasedBound + time.Minute)
|
|
return v
|
|
}
|
|
|
|
func running(node, module, fp string, stopped bool) (string, []byte) {
|
|
data, _ := json.Marshal(nodeengine.Running{Node: node, Module: module, Fingerprint: fp, Stopped: stopped, At: time.Unix(1, 0)})
|
|
return nodeengine.SubjectRunning(node, module), data
|
|
}
|
|
|
|
// TestRepublish: design 47 G §588 (A-73, A-76): a reserved version left unpublished is published again, on a registry
|
|
// holder's running and on its node's returned, once per message by its stream sequence, as a publish by digest with no parts,
|
|
// at most the setting's number per event, newest first; an artifact no store holds is said once and not tried again.
|
|
func TestRepublish(t *testing.T) {
|
|
ctx := context.Background()
|
|
t.Run("running of a holder, once per sequence", func(t *testing.T) {
|
|
b := newRepublishBed(t, 50)
|
|
v := b.reserve(t, "m")
|
|
if v.Fingerprint != "src2:m" {
|
|
t.Fatalf("the reservation keeps the entry's fingerprint: %q", v.Fingerprint)
|
|
}
|
|
subject, data := running("n1", "oci", "src2:holder", false)
|
|
for _, seq := range []uint64{5, 4} { // 4 is older than the acted 5: it starts nothing
|
|
if _, err := b.Heard(ctx, subject, data, seq); err != nil || len(b.sent) != 1 {
|
|
t.Fatalf("seq %d: sent %d (%v), want 1", seq, len(b.sent), err)
|
|
}
|
|
}
|
|
p := b.sent[0]
|
|
if p.Seat != SeatArtifacts || p.Name != "m/prog" || p.Kind != "bundle" || p.Version != v.Version || p.Digest != "sha256:aa" || p.Fingerprint != "src2:m" {
|
|
t.Errorf("sent %+v, want the bundle by digest to the oci seat, the Go package left to its own seat's holder", p)
|
|
}
|
|
if _, err := b.Heard(ctx, subject, data, 5); err != nil || len(b.sent) != 1 {
|
|
t.Errorf("the same sequence read again sent %d, want still 1", len(b.sent))
|
|
}
|
|
if _, err := b.Heard(ctx, subject, data, 6); err != nil || len(b.sent) != 2 {
|
|
t.Errorf("a newer sequence sent %d, want 2", len(b.sent))
|
|
}
|
|
if keys, _ := b.queue.KeysOf(ctx, AllFilter); len(keys) != 0 {
|
|
t.Errorf("a key stays in the queue after its outcome: %v", keys)
|
|
}
|
|
})
|
|
t.Run("returned acts, started does not", func(t *testing.T) {
|
|
b := newRepublishBed(t, 50)
|
|
b.reserve(t, "m")
|
|
subject, data := running("n1", "oci", "src2:holder", true) // stopped: not a holder yet
|
|
_, _ = b.Heard(ctx, subject, data, 1)
|
|
subject, data = running("n1", "oci", "src2:holder", false)
|
|
_, _ = b.Heard(ctx, subject, data, 2)
|
|
b.sent = nil
|
|
for i, why := range []string{nodeengine.ManifestReadsChangedWhyStarted, "", nodeengine.ManifestReadsChangedWhyReturned} {
|
|
ev, _ := json.Marshal(nodeengine.ManifestReadsChanged{Node: "n1", Why: why})
|
|
handled, err := b.Heard(ctx, nodeengine.SubjectManifestReadsChanged("n1"), ev, uint64(10+i))
|
|
if !handled || err != nil {
|
|
t.Fatalf("%q: handled %v, %v", why, handled, err)
|
|
}
|
|
if want := map[string]int{"started": 0, "": 0, "returned": 1}[why]; len(b.sent) != want {
|
|
t.Errorf("why %q: sent %d, want %d", why, len(b.sent), want)
|
|
}
|
|
}
|
|
if handled, _ := b.Heard(ctx, "mesh.seat.node-engine.event.composed.n1", nil, 1); handled {
|
|
t.Error("a subject that is neither is taken as its own")
|
|
}
|
|
})
|
|
t.Run("the setting limits an event, newest first", func(t *testing.T) {
|
|
b := newRepublishBed(t, 2)
|
|
var versions []string
|
|
for _, m := range []string{"m1", "m2", "m3"} {
|
|
versions = append(versions, b.reserve(t, m).Module)
|
|
}
|
|
subject, data := running("n1", "oci", "src2:holder", false)
|
|
if _, err := b.Heard(ctx, subject, data, 1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(b.sent) != 2 || !strings.HasPrefix(b.sent[0].Name, "m3/") || !strings.HasPrefix(b.sent[1].Name, "m2/") {
|
|
t.Errorf("sent %+v, want the two newest, m3 then m2", b.sent)
|
|
}
|
|
})
|
|
t.Run("an artifact no store holds is said once and not tried again", func(t *testing.T) {
|
|
b := newRepublishBed(t, 50)
|
|
b.reserve(t, "m")
|
|
b.refuse = "oci-artifact-registry's publish of m/prog: not stored: no blob sha256:aa"
|
|
subject, data := running("n1", "oci", "src2:holder", false)
|
|
_, _ = b.Heard(ctx, subject, data, 1)
|
|
_, _ = b.Heard(ctx, subject, data, 2)
|
|
if len(b.sent) != 1 || len(b.said) != 1 || !strings.Contains(b.said[0], "not stored: ") {
|
|
t.Errorf("sent %d, said %v: want one try and one saying of not stored", len(b.sent), b.said)
|
|
}
|
|
})
|
|
t.Run("a refusal in other words is said each time", func(t *testing.T) {
|
|
b := newRepublishBed(t, 50)
|
|
b.reserve(t, "m")
|
|
b.refuse = "the holder is full"
|
|
subject, data := running("n1", "oci", "src2:holder", false)
|
|
_, _ = b.Heard(ctx, subject, data, 1)
|
|
_, _ = b.Heard(ctx, subject, data, 2)
|
|
if len(b.sent) != 2 || len(b.said) != 2 {
|
|
t.Errorf("sent %d, said %v: want both events tried and said", len(b.sent), b.said)
|
|
}
|
|
})
|
|
t.Run("a queue past its cap says queue full", func(t *testing.T) {
|
|
b := newRepublishBed(t, 50)
|
|
b.reserve(t, "m")
|
|
b.queue.bytes = PublishQueueMiB << 20
|
|
subject, data := running("n1", "oci", "src2:holder", false)
|
|
_, _ = b.Heard(ctx, subject, data, 1)
|
|
if len(b.sent) != 0 || len(b.said) != 1 || !strings.HasSuffix(b.said[0], QueueFull) {
|
|
t.Errorf("sent %d, said %v, want nothing sent and queue full said", len(b.sent), b.said)
|
|
}
|
|
})
|
|
t.Run("a restart sends the queued again, a begun one anew", func(t *testing.T) {
|
|
b := newRepublishBed(t, 50)
|
|
v := b.reserve(t, "m")
|
|
raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: "oci", Module: "m", Version: v.Version, Artifacts: map[string]string{"m:prog": "sha256:aa"}, Queued: b.now, Begun: true})
|
|
if _, err := b.queue.Create(ctx, QueueKey("oci", "m", v.Version), raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := b.Resume(ctx); err != nil || len(b.sent) != 1 {
|
|
t.Errorf("resume sent %d (%v), want the begun version sent anew", len(b.sent), err)
|
|
}
|
|
})
|
|
t.Run("the tool queues for a holder known to run", func(t *testing.T) {
|
|
b := newRepublishBed(t, 1)
|
|
b.reserve(t, "m1")
|
|
b.reserve(t, "m2")
|
|
if q, err := b.Republish(ctx, "", ""); err != nil || len(q) != 0 {
|
|
t.Errorf("no holder known: queued %v (%v), want none", q, err)
|
|
}
|
|
subject, data := running("n1", "oci", "src2:holder", false)
|
|
_, _ = b.Heard(ctx, subject, data, 1) // Max 1: only m2 sent
|
|
b.sent = nil
|
|
q, err := b.Republish(ctx, "", "")
|
|
if err != nil || fmt.Sprint(q) != "[m2 0.1.0 m1 0.1.0]" || len(b.sent) != 2 {
|
|
t.Errorf("the tool with neither queued %v (%v), sent %d: want both versions, with no limit", q, err, len(b.sent))
|
|
}
|
|
if _, err := b.Republish(ctx, "m1", "0.1.9"); err == nil {
|
|
t.Error("a version no reservation holds is not refused")
|
|
}
|
|
answer, err := AnswerQueue(Reader{Keys: func() ([]string, error) { return nil, nil }}, Reader{Keys: func() ([]string, error) { return b.acted.KeysOf(ctx, AllFilter) },
|
|
Get: func(k string) (json.RawMessage, bool, error) { v, _, ok, err := b.acted.Get(ctx, k); return v, ok, err }})
|
|
if err != nil || answer.Acted[ActedKey(subject)] != 1 {
|
|
t.Errorf("builder_publish_queue acted = %v (%v), want the running's sequence 1", answer.Acted, err)
|
|
}
|
|
})
|
|
}
|