Merge branch 'feat/47-G-build' into feat/47-G-build-teller

This commit is contained in:
2026-11-11 11:53:45 +01:00
40 changed files with 2677 additions and 46 deletions
@@ -6,3 +6,12 @@ require (
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
)
require (
github.com/klauspost/compress v1.20.0 // indirect
github.com/nats-io/nats.go v1.54.0 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/sys v0.48.0 // indirect
)
+20 -6
View File
@@ -1,12 +1,26 @@
github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
@@ -0,0 +1,200 @@
package meshcatalogclient
import (
"context"
"encoding/json"
"fmt"
"slices"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/views"
)
// RegistrationsVerb is the seat verb a part takes its first view of the registrations by, as its manifest's invokes
// names it.
const RegistrationsVerb = "seat:" + seat.Seat + "." + seat.VerbRegistrations
// Call asks one seat verb over the bus and answers what the holder said, whatever wraps it (process.Conn.Call, or
// stdio.Ask through the tool runner).
type Call func(key string, body any) (json.RawMessage, error)
// registrationsCreated stands for the picture's Created in its follower: the registrations are followed by the seat's
// event, never by an owner's views family, so one fixed time, never zero.
var registrationsCreated = time.Unix(0, 0).UTC()
// Registrations is a part's typed follower of the registrations (design 47 G §353): a first view asked of the seat's verb
// `registrations`, then each `registration-changed` heard, built on go/views' Follower, so the runtime's tools
// `<module>_first_views` and `<module>_first_view_again` read and steer it as every first view. Its picture is keyed by
// module, each value the module's newest registration. Ready is closed at the first view: a part that switches on its
// first registrations waits for it.
type Registrations struct {
Follower *views.Follower
call Call
}
// NewRegistrations is a Registrations asking the seat through call. Nothing is asked before Start.
func NewRegistrations(call Call) *Registrations {
r := &Registrations{call: call}
r.Follower = &views.Follower{Owner: seat.Seat, Tool: RegistrationsVerb, State: seat.StateRegistrations, Ask: r.ask}
return r
}
// WireRegistrations is a Registrations adopted by a stdio part's views.Reader: the reader writes its conditions, reads it
// as `mesh-catalog.registrations`, lists and asks it again in its first-view tools, and at its Start subscribes the
// seat's event registration-changed (the part's manifest consumes it) into Hear and asks the first view. changed, when
// not nil, is told of each module whose registration the picture changes, after the reader's own. The part declares the
// seat's verb registrations in invokes.
func WireRegistrations(r *views.Reader, call Call, changed func(reg seat.Registration, gone bool)) (*Registrations, error) {
g := NewRegistrations(call)
g.Follower.Changed = typedChanged(changed)
if err := r.Adopt(g.Follower, seat.Seat, seat.EventRegistrationChanged, g.Hear); err != nil {
return nil, err
}
return g, nil
}
// WireRegistrationsProcess is WireRegistrations for a process's views.ProcessReader: its picture
// (`mesh-catalog_registrations`) reads the registrations as a read-only bucket, and the seat's event
// registration-changed on the module's durable consumer goes to Hear.
func WireRegistrationsProcess(r *views.ProcessReader, call Call, changed func(reg seat.Registration, gone bool)) (*Registrations, *views.Picture, error) {
g := NewRegistrations(call)
g.Follower.Changed = typedChanged(changed)
p, err := r.Adopt(g.Follower, seat.Seat, seat.EventRegistrationChanged, g.Hear)
if err != nil {
return nil, nil, err
}
return g, p, nil
}
func typedChanged(changed func(seat.Registration, bool)) func(string, json.RawMessage, bool, time.Time) {
if changed == nil {
return nil
}
return func(module string, value json.RawMessage, gone bool, _ time.Time) {
reg := seat.Registration{Module: module}
if !gone {
_ = json.Unmarshal(value, &reg)
}
changed(reg, gone)
}
}
// Start asks the first view until it is answered or ctx ends.
func (r *Registrations) Start(ctx context.Context) error { return r.Follower.Start(ctx) }
// Ready is closed once the first view is answered.
func (r *Registrations) Ready() <-chan struct{} { return r.Follower.Ready() }
// Registration is one module's newest registration in the picture.
func (r *Registrations) Registration(module string) (seat.Registration, bool) {
raw, ok := r.Follower.Get(module)
if !ok {
return seat.Registration{}, false
}
var reg seat.Registration
if err := json.Unmarshal(raw, &reg); err != nil {
return seat.Registration{}, false
}
return reg, true
}
// All is every registration in the picture, by module name.
func (r *Registrations) All() []seat.Registration {
values := r.Follower.Values()
out := make([]seat.Registration, 0, len(values))
for _, raw := range values {
var reg seat.Registration
if json.Unmarshal(raw, &reg) == nil {
out = append(out, reg)
}
}
slices.SortFunc(out, func(a, b seat.Registration) int {
if a.Module < b.Module {
return -1
}
if a.Module > b.Module {
return 1
}
return 0
})
return out
}
// Hear takes one `registration-changed` event, as the part's durable consumer of the seat's events hands it. A version
// the picture holds, or one older, is passed over; a newer one asks the first view again on its own goroutine, so the
// consumer is never held while the holder is away, and again until the picture holds that version or ctx ends (an ask
// already running, the first view among them, may have been answered before the change was written). Data that is no such event is an error.
func (r *Registrations) Hear(ctx context.Context, data []byte) error {
var ev seat.RegistrationChangedEvent
if err := json.Unmarshal(data, &ev); err != nil || ev.Module == "" || ev.Version <= 0 {
return fmt.Errorf("meshcatalogclient: not a %s event: %s", seat.EventRegistrationChanged, clipBytes(data))
}
if r.holds(ev.Module, ev.Version) {
return nil
}
go r.catchUp(ctx, ev.Module, ev.Version)
return nil
}
func (r *Registrations) holds(module string, version int64) bool {
reg, ok := r.Registration(module)
return ok && reg.Version >= version
}
// catchUp asks the first view again until the picture holds version of module, pausing between asks as the follower's
// Wait says (doubling to a minute), or until ctx ends.
func (r *Registrations) catchUp(ctx context.Context, module string, version int64) {
wait := r.Follower.Wait
if wait == 0 {
wait = 5 * time.Second
}
for !r.holds(module, version) {
if err := r.Follower.Gap(ctx, fmt.Sprintf("registration of %s at version %d said", module, version)); err != nil {
return
}
if r.holds(module, version) {
return
}
p, cancel := context.WithTimeout(ctx, wait)
<-p.Done()
cancel()
if ctx.Err() != nil {
return
}
if wait *= 2; wait > time.Minute {
wait = time.Minute
}
}
}
// ask is the Follower's Ask: every registration, whole in one answer.
func (r *Registrations) ask(_ context.Context, _ string) (views.Answer, error) {
raw, err := r.call(RegistrationsVerb, seat.RegistrationsInput{})
if err != nil {
return views.Answer{}, err
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return views.Answer{}, fmt.Errorf("%s answered: %s", RegistrationsVerb, said)
}
var a seat.RegistrationsAnswer
if err := json.Unmarshal(answer, &a); err != nil || a.Registrations == nil {
return views.Answer{}, fmt.Errorf("%s answered no registrations: %s", RegistrationsVerb, clipBytes(answer))
}
values := make(map[string]json.RawMessage, len(a.Registrations))
for _, reg := range a.Registrations {
v, _ := json.Marshal(reg)
values[reg.Module] = v
}
return views.Answer{V: 1, Owner: seat.Seat, State: seat.StateRegistrations, Created: registrationsCreated, Values: values, At: time.Now()}, nil
}
func clipBytes(b []byte) string {
if len(b) > 200 {
return string(b[:200]) + "…"
}
return string(b)
}
@@ -0,0 +1,140 @@
package meshcatalogclient
import (
"context"
"encoding/json"
"fmt"
"sync"
"testing"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// fakeRegistrations answers the verb registrations from the versions it keeps, and counts each ask.
type fakeRegistrations struct {
mu sync.Mutex
versions map[string]int64
asks int
// stale answers the next ask from these versions, once: an answer read before a change was written.
stale map[string]int64
}
func (f *fakeRegistrations) call(key string, body any) (json.RawMessage, error) {
if key != RegistrationsVerb {
return nil, fmt.Errorf("asked %s", key)
}
if in := body.(seat.RegistrationsInput); in.Module != "" || in.History {
return nil, fmt.Errorf("asked %+v", in)
}
f.mu.Lock()
defer f.mu.Unlock()
f.asks++
versions := f.versions
if f.stale != nil {
versions, f.stale = f.stale, nil
}
a := seat.RegistrationsAnswer{Registrations: []seat.Registration{}}
for m, v := range versions {
a.Registrations = append(a.Registrations, seat.Registration{Module: m, Version: v, Origin: "novox/" + m, Path: "module.json", Branches: []string{"main"}})
}
return json.Marshal(a)
}
func (f *fakeRegistrations) set(module string, version int64) {
f.mu.Lock()
defer f.mu.Unlock()
f.versions[module] = version
}
func (f *fakeRegistrations) asked() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.asks
}
func changedEvent(module string, version int64) []byte {
b, _ := json.Marshal(seat.RegistrationChangedEvent{V: 1, Module: module, Version: version, Moved: []string{"branches"}, Warrant: "ask-1"})
return b
}
// CT10 (design 47 G §353): a part following the registrations takes a first view of the verb registrations, then each
// registration-changed: a newer version is asked and held, one it holds asks nothing, a change heard while an answer
// read before it was given asks again, and what is no such event is refused.
func TestRegistrationsAreAFirstViewThenTheirEvents(t *testing.T) {
changed := make(chan seat.Registration, 16)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
next := func() seat.Registration {
t.Helper()
select {
case reg := <-changed:
return reg
case <-ctx.Done():
t.Fatal("no registration changed")
return seat.Registration{}
}
}
f := &fakeRegistrations{versions: map[string]int64{"nats": 1, "zsh": 3}}
r := NewRegistrations(f.call)
r.Follower.Wait = time.Nanosecond
r.Follower.Changed = typedChanged(func(reg seat.Registration, gone bool) {
if !gone {
changed <- reg
}
})
select {
case <-r.Ready():
t.Fatal("ready before its first view")
default:
}
if err := r.Start(ctx); err != nil {
t.Fatal(err)
}
<-r.Ready()
if reg, ok := r.Registration("nats"); !ok || reg.Version != 1 || reg.Origin != "novox/nats" {
t.Fatalf("first view holds nats as %+v, %v", reg, ok)
}
if all := r.All(); len(all) != 2 || all[0].Module != "nats" || all[1].Module != "zsh" {
t.Fatalf("All = %+v", all)
}
for range 2 {
next()
}
if err := r.Hear(ctx, changedEvent("zsh", 3)); err != nil {
t.Fatal(err)
}
if n := f.asked(); n != 1 {
t.Errorf("a version held asked the first view again: %d asks", n)
}
f.set("nats", 2)
if err := r.Hear(ctx, changedEvent("nats", 2)); err != nil {
t.Fatal(err)
}
if reg := next(); reg.Module != "nats" || reg.Version != 2 {
t.Fatalf("after registration-changed: %+v", reg)
}
// The answer to the ask the event starts was read before the change was written: it is asked once more.
f.mu.Lock()
f.stale = map[string]int64{"nats": 2, "zsh": 3}
f.mu.Unlock()
f.set("zsh", 4)
if err := r.Hear(ctx, changedEvent("zsh", 4)); err != nil {
t.Fatal(err)
}
if reg := next(); reg.Module != "zsh" || reg.Version != 4 {
t.Fatalf("after a stale answer: %+v", reg)
}
if n := f.asked(); n != 4 {
t.Errorf("asks = %d, want 4 (first view, nats, a stale answer and the one after)", n)
}
for _, data := range []string{`{}`, `not json`, `{"module":"nats"}`} {
if err := r.Hear(ctx, []byte(data)); err == nil {
t.Errorf("Hear(%s) took it", data)
}
}
}
@@ -0,0 +1,176 @@
package meshwantsclient
import (
"context"
"encoding/json"
"fmt"
"strconv"
"sync"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/views"
)
// RevisionVerb is the seat verb a part takes its first view of the assignment half by, as its manifest's invokes names
// it.
const RevisionVerb = "seat:" + seat.Seat + "." + seat.VerbRevision
// kindTrunk is the kind of an assignment graph revision a running part acts on; a candidate's moves nothing.
const kindTrunk = "trunk"
// Call asks one seat verb over the bus and answers what the holder said, whatever wraps it (process.Conn.Call, or
// stdio.Ask through the tool runner).
type Call func(key string, body any) (json.RawMessage, error)
// assignmentCreated stands for the view's Created in its follower: a revision view is never made again, so one fixed
// time, never zero, lets the follower tell a revision heard from one it holds.
var assignmentCreated = time.Unix(0, 0).UTC()
// assignmentViewName is how an assignment half's revision is named in a part's first-view status (views.Status.View,
// WaitsFor).
func assignmentViewName(revision int64) string { return strconv.FormatInt(revision, 10) }
// Assignment is a part's view of the assignment half (design 47 G §245): a first view of the newest trunk revision asked
// of the seat mesh-wants' verb `revision`, then kept from its event `assignment-revision`, built on go/views' Follower,
// so the runtime's tools `<module>_first_views` and `<module>_first_view_again` read and steer it as every first view.
// Its picture is keyed `<node>.<module>`, each value that entry's reference. With no view the part acts on nothing:
// Revision says false, and a revision heard and not held yet is waited for, said in the status, and past
// views.WaitsBound raised as view-waits.
type Assignment struct {
Follower *views.Follower
// Held, when set, is told of each revision once the view holds it, on the ask's goroutine and outside any lock.
Held func(seat.AssignmentRevision)
call Call
mu sync.Mutex
held *seat.AssignmentRevision
waits int64
}
// NewAssignment is an Assignment asking the seat through call. Nothing is asked before Start.
func NewAssignment(call Call) *Assignment {
a := &Assignment{call: call}
a.Follower = &views.Follower{Owner: seat.Seat, Tool: RevisionVerb, State: seat.StateAssignmentRevisions, Ask: a.ask}
return a
}
// WireAssignment is an Assignment adopted by a stdio part's views.Reader: the reader writes its conditions and its
// view-waits finding, reads it as `mesh-wants.assignment-revisions`, lists and asks it again in its first-view tools, and
// at its Start subscribes the seat's event assignment-revision (the part's manifest consumes it) into Hear and asks the
// first view. held, when not nil, is the Assignment's Held. The part declares the seat's verb revision in invokes.
func WireAssignment(r *views.Reader, call Call, held func(seat.AssignmentRevision)) (*Assignment, error) {
a := NewAssignment(call)
a.Held = held
if err := r.Adopt(a.Follower, seat.Seat, seat.EventAssignmentRevision, a.Hear); err != nil {
return nil, err
}
return a, nil
}
// WireAssignmentProcess is WireAssignment for a process's views.ProcessReader: its picture
// (`mesh-wants_assignment-revisions`) reads the half as a read-only bucket, and the seat's event assignment-revision on
// the module's durable consumer goes to Hear.
func WireAssignmentProcess(r *views.ProcessReader, call Call, held func(seat.AssignmentRevision)) (*Assignment, *views.Picture, error) {
a := NewAssignment(call)
a.Held = held
p, err := r.Adopt(a.Follower, seat.Seat, seat.EventAssignmentRevision, a.Hear)
if err != nil {
return nil, nil, err
}
return a, p, nil
}
// Start asks the first view, the newest trunk revision, until it is answered or ctx ends.
func (a *Assignment) Start(ctx context.Context) error { return a.Follower.Start(ctx) }
// Revision is the assignment graph revision the view holds; false while it has none, when the part acts on nothing new.
func (a *Assignment) Revision() (seat.AssignmentRevision, bool) {
a.mu.Lock()
defer a.mu.Unlock()
if a.held == nil {
return seat.AssignmentRevision{}, false
}
return *a.held, true
}
// Hear takes one `assignment-revision` event, as the part's durable consumer of the seat's events hands it. A
// candidate's revision moves nothing; a trunk revision the view holds, or one older, is passed over; a newer one is
// waited for and asked on its own goroutine, so the consumer is never held while the holder is away. Data that is no
// such event is an error.
func (a *Assignment) Hear(ctx context.Context, data []byte) error {
var ev seat.AssignmentRevisionEvent
if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 {
return fmt.Errorf("meshwantsclient: not a %s event: %s", seat.EventAssignmentRevision, clipBytes(data))
}
if ev.Kind != kindTrunk {
return nil
}
a.mu.Lock()
held := int64(0)
if a.held != nil {
held = a.held.Revision
}
if ev.Revision > a.waits {
a.waits = ev.Revision
}
a.mu.Unlock()
if ev.Revision <= held {
return nil
}
// The follower keeps the newest revision heard, so an ask that was running when it was heard is asked once more.
a.Follower.Apply(views.Event{Owner: seat.Seat, State: seat.StateAssignmentRevisions, Revision: uint64(ev.Revision), Created: assignmentCreated, Gap: true})
a.Follower.WaitFor(ctx, assignmentViewName(ev.Revision))
go func() {
_ = a.Follower.Gap(ctx, fmt.Sprintf("assignment revision %d said, the view holds %d", ev.Revision, held))
}()
return nil
}
// ask is the Follower's Ask: the newest trunk revision, whole in one answer (its entries are named by digest).
func (a *Assignment) ask(_ context.Context, _ string) (views.Answer, error) {
raw, err := a.call(RevisionVerb, seat.RevisionInput{})
if err != nil {
return views.Answer{}, err
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return views.Answer{}, fmt.Errorf("%s answered: %s", RevisionVerb, said)
}
var ans seat.RevisionAnswer
if err := json.Unmarshal(answer, &ans); err != nil {
return views.Answer{}, fmt.Errorf("%s answered what is not its answer: %s", RevisionVerb, clipBytes(answer))
}
r := ans.Revision
if r.Revision <= 0 || r.Kind != kindTrunk {
return views.Answer{}, fmt.Errorf("%s answered a %q revision %d for the trunk", RevisionVerb, r.Kind, r.Revision)
}
values := make(map[string]json.RawMessage, len(r.Entries))
for _, e := range r.Entries {
v, _ := json.Marshal(e)
values[e.Node+"."+e.Module] = v
}
a.mu.Lock()
moved := a.held == nil || a.held.Revision != r.Revision
a.held = &r
holds := a.waits <= r.Revision
if holds {
a.waits = 0
}
a.mu.Unlock()
if holds {
a.Follower.Holds(assignmentViewName(r.Revision))
}
if moved && a.Held != nil {
a.Held(r)
}
return views.Answer{V: 1, Owner: seat.Seat, State: seat.StateAssignmentRevisions, Revision: uint64(r.Revision), Created: assignmentCreated, Values: values, At: time.Now()}, nil
}
func clipBytes(b []byte) string {
if len(b) > 200 {
return string(b[:200]) + "…"
}
return string(b)
}
@@ -0,0 +1,116 @@
package meshwantsclient
import (
"context"
"encoding/json"
"fmt"
"sync"
"testing"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
)
// fakeWants answers the verb revision with its newest trunk revision, and counts each ask.
type fakeWants struct {
mu sync.Mutex
newest int64
asks int
}
func (f *fakeWants) call(key string, body any) (json.RawMessage, error) {
if key != RevisionVerb {
return nil, fmt.Errorf("asked %s", key)
}
if in := body.(seat.RevisionInput); in != (seat.RevisionInput{}) {
return nil, fmt.Errorf("asked %+v", in)
}
f.mu.Lock()
defer f.mu.Unlock()
f.asks++
r := seat.AssignmentRevision{Revision: f.newest, Kind: kindTrunk, Placements: "p",
Catalogs: []seat.CatalogOf{{Branch: "main", Revision: 10 + f.newest}},
Entries: []seat.NodeEntryRef{{Node: "anchor", Module: "nats", Digest: fmt.Sprintf("d%d", f.newest)}}}
return json.Marshal(seat.RevisionAnswer{Revision: r})
}
func (f *fakeWants) set(newest int64) {
f.mu.Lock()
defer f.mu.Unlock()
f.newest = newest
}
func (f *fakeWants) asked() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.asks
}
func revisionEvent(rev int64, kind string) []byte {
b, _ := json.Marshal(seat.AssignmentRevisionEvent{V: 1, Revision: rev, Kind: kind})
return b
}
// CT10 (design 47 G §245): a part takes a first view of the assignment half by the verb revision, then follows its
// event assignment-revision: a newer trunk revision is waited for, said in the status, and held; a candidate's, or one
// held, asks nothing; what is no such event is refused.
func TestTheAssignmentHalfIsAFirstViewThenItsEvents(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
held := make(chan seat.AssignmentRevision, 8)
next := func() seat.AssignmentRevision {
t.Helper()
select {
case r := <-held:
return r
case <-ctx.Done():
t.Fatal("no revision held")
return seat.AssignmentRevision{}
}
}
f := &fakeWants{newest: 3}
a := NewAssignment(f.call)
a.Follower.Wait = time.Nanosecond
a.Held = func(r seat.AssignmentRevision) { held <- r }
if _, ok := a.Revision(); ok {
t.Fatal("a revision before its first view")
}
if err := a.Start(ctx); err != nil {
t.Fatal(err)
}
if r := next(); r.Revision != 3 || len(r.Catalogs) != 1 || r.Catalogs[0].Revision != 13 {
t.Fatalf("first view holds %+v", r)
}
if v, ok := a.Follower.Get("anchor.nats"); !ok || string(v) == "" {
t.Errorf("the picture lacks anchor.nats: %s", v)
}
for _, ev := range [][]byte{revisionEvent(3, kindTrunk), revisionEvent(2, kindTrunk), revisionEvent(9, "candidate")} {
if err := a.Hear(ctx, ev); err != nil {
t.Fatal(err)
}
}
if n := f.asked(); n != 1 {
t.Errorf("a revision held, an older one or a candidate's asked again: %d asks", n)
}
f.set(4)
if err := a.Hear(ctx, revisionEvent(4, kindTrunk)); err != nil {
t.Fatal(err)
}
if r := next(); r.Revision != 4 {
t.Fatalf("after assignment-revision 4 the view holds %d", r.Revision)
}
if s := a.Follower.Status(); s.View != "4" || s.WaitsFor != "" {
t.Errorf("status view %q, waits for %q", s.View, s.WaitsFor)
}
if r, ok := a.Revision(); !ok || r.Revision != 4 {
t.Errorf("Revision = %d, %v", r.Revision, ok)
}
for _, data := range []string{`{}`, `not json`, `{"kind":"trunk"}`} {
if err := a.Hear(ctx, []byte(data)); err == nil {
t.Errorf("Hear(%s) took it", data)
}
}
}
@@ -6,3 +6,12 @@ require (
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
)
require (
github.com/klauspost/compress v1.20.0 // indirect
github.com/nats-io/nats.go v1.54.0 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/sys v0.48.0 // indirect
)
@@ -0,0 +1,24 @@
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
@@ -79,7 +79,7 @@ var readFamilies = []string{buckets.ComposedRead}
var ownBuckets = []string{buckets.Asks, buckets.Differences,
buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket,
plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket,
engineReadsBucket, retiredBucket, copiesBucket}
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket}
// engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key `<node>`, as
// the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again.
@@ -88,6 +88,9 @@ const engineReadsState = "engine-reads"
// engineReadsBucket is that state's bucket.
const engineReadsBucket = assignment.Module + "_" + engineReadsState
// writtenBucket is the bucket of the state written (pipelines.go): the pipeline ends the wants follow.
const writtenBucket = assignment.Module + "_" + writtenState
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
var seatBuckets = []string{plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket,
@@ -115,6 +118,7 @@ var keptStates = []announce.State{
{Name: engineReadsState, Readable: true},
{Name: retiredState}, // read only by the copy into assignments (migrate.go)
{Name: copiesState, Readable: true},
{Name: writtenState, Readable: true},
}
// kvSet is every bucket the assigner opened, by name.
@@ -197,7 +201,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
func startsWithout(name string) bool {
switch name {
case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket,
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket:
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket:
return true
}
return slices.Contains(seatBuckets, name) || name == buckets.Grants
@@ -297,6 +301,7 @@ var watched = []string{
buckets.Source,
buckets.Merges,
engineReadsBucket,
writtenBucket,
}
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
@@ -485,6 +490,14 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
logger.Printf("%s is not heard, so an ask that ends at the operator channel without an answer is not asked again until the assigner starts again: %v", askEndedSubject, err)
}
readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], staleOfWorld(w), bus, callWait{heard: func() bool { return w.seatHeard(oc.Seat) }, findings: findings}, logger.Printf)
if kv := kvs[writtenBucket]; kv != nil {
// The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow
// once the first registrations are written (§95, §353).
p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf)
if err := hearEnds(hub, p.heard); err != nil {
logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err)
}
}
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
}
if bus != nil && kvs[buckets.Puts] != nil {
@@ -58,6 +58,10 @@ type passer struct {
// neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the
// first pass after a start, which clears what a previous process may have raised (startClears).
neither map[string]neitherOne
// ends are the pipeline ends the wants follow (pipelines.go); nil is the trunk-only want, kept until the first registrations.
ends *pipelineEnds
// heard is the same follower while the wants do not follow it yet (ends is nil).
heard *pipelineEnds
// acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket).
acts *putActs
// noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is
@@ -267,9 +271,16 @@ func (p *passer) wants(ctx context.Context) {
moduleGraph := p.moduleGraph()
served := p.served(ctx, get, moduleGraph, nodes)
offering := p.offering(ctx, get, manifests, nodes) // bus state (ADR 0614 §5)
assignedBranches := p.w.placedBranches()
for name := range names {
fp := trunk[name].Fp2
name := name
// A want follows a passed pipeline: a module whose fingerprint no pipeline passed at on every branch its placements
// chose, and whose wants do not name it already, gets none (design 47 G §29, §35).
branches := p.ends.branchesOf(name, assignedBranches)
if fp != "" && !p.ends.mayWant(name, fp, branches, wanted[name]) {
continue
}
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp),
Self: func(node string) bool { return placed[node] },
Composed: func(node, fp string) (assign.ComposedDigest, bool) {
@@ -359,6 +370,9 @@ func (p *passer) wants(ctx context.Context) {
want := want
p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeWant(want) })
}
if len(out.Wants) > 0 {
p.ends.wrote(name, fp, branches)
}
}
}
@@ -1128,10 +1128,10 @@ func aReleasesApproveAnswersTheMoveOfItsBuildAlone(t *testing.T) {
p.logf = t.Logf
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: commit, Src: 7, FpSrc: c.fpSrc, Path: meshcatalogclient.RootPath})
p.w.put(buckets.Modules, "web", mod)
src, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}})
p.w.putEntry(buckets.Source, mdc.SourceMainKey, src, 7, time.Now())
later, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}})
p.w.putEntry(buckets.Source, mdc.SourceMainKey, later, 8, time.Now()) // a later merge leaves revision 7's release
src, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}})
p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), src, 7, time.Now())
later, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}})
p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), later, 8, time.Now()) // a later merge leaves revision 7's release
if c.mergedAt != "" {
merge, err := mdc.EncodeMerge(mdseat.MergesByGroupValue{Group: "feat/web", Ask: "merge.feat_web.0123456789ab", Pulls: []mdseat.MergesByGroupValuePullsItem{
{Repository: "novox/web", Number: 3, Head: strings.Repeat("3", 40), State: string(mdc.MergeMerged), Commit: c.mergedAt}}})
@@ -0,0 +1,327 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"sync"
"github.com/nats-io/nats.go/jetstream"
mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient"
mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/process"
)
// The wants follow a passed pipeline (design 47 G §29, §35, §325, §352): the runner says each pipeline's end as
// pipeline-ended, and a group's as group-ended. The assigner keeps, in its old state `written`, one key per module and
// watched branch: the source revision and end count of the newest end it heard for it, and whether the wants were written
// from it. A newer end replaces it by compare-and-set; an older one, or the same one again, writes nothing.
// writtenState is the assigner's state of the pipeline ends its wants follow; writtenTool reads it.
const (
writtenState = "written"
writtenTool = "assigner_written"
)
// pipelineEndedSubject and groupEndedSubject are the runner's events the assigner consumes.
const (
pipelineEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventPipelineEnded + ".>"
groupEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventGroupEnded + ".>"
)
// End is one value of the state written: the newest end of a module's pipeline on a watched branch.
type End struct {
Source int64 `json:"source"`
EndCount int64 `json:"end-count"`
Fp string `json:"fp"`
Commit string `json:"commit,omitempty"`
// Passed: the pipeline ended passed, every required step passed (and, for a group, the group ended passed).
Passed bool `json:"passed"`
// Written: the wants were written from this end.
Written bool `json:"written,omitempty"`
Why string `json:"why,omitempty"`
}
// endKey is the key of a module on a watched branch, each name escaped as a key's name is.
func endKey(module, branch string) string {
return keys.EscapeName(module) + "." + keys.EscapeName(branch)
}
// newer says whether an end of (source, count) replaces the one kept: a later source revision, or the same one ended again.
func (e End) newer(source, count int64) bool {
return source > e.Source || (source == e.Source && count > e.EndCount)
}
// requiredPassed says whether no required step of a pipeline stands failed, unasked or unended (design 47 G §30: a failed
// optional step holds nothing).
func requiredPassed(p mpr.PipelineEnded) bool {
for _, s := range p.Steps {
if s.Status != mpr.StepResultStatusRequired {
continue
}
switch s.Outcome {
case mpr.StepResultOutcomeFailed, mpr.StepResultOutcomeCouldNotRun, mpr.StepResultOutcomeDeclined, mpr.StepResultOutcomeNotAsked,
mpr.StepResultOutcomeCancelled, mpr.StepResultOutcomeSuperseded, mpr.StepResultOutcomeDidNotBuild,
mpr.StepResultOutcomeRunning, mpr.StepResultOutcomeWaiting, mpr.StepResultOutcomeWaitingForABuildNode:
return false
}
}
return p.Outcome == mpr.PipelineEndedOutcomePassed
}
// endStore is the kept state written: the bus bucket, or a test's.
type endStore interface {
// get is a key's value and revision; none for a key not there.
get(key string) (value []byte, revision uint64, err error)
// swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not.
swap(key string, value []byte, revision uint64) error
}
var errSwapped = errors.New("the key changed")
// kvEnds is endStore over a bus bucket.
type kvEnds struct {
ctx context.Context
kv jetstream.KeyValue
}
func (k kvEnds) get(key string) ([]byte, uint64, error) {
e, err := k.kv.Get(k.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, 0, nil
}
if err != nil {
return nil, 0, err
}
return e.Value(), e.Revision(), nil
}
func (k kvEnds) swap(key string, value []byte, revision uint64) error {
var err error
if revision == 0 {
_, err = k.kv.Create(k.ctx, key, value)
} else {
_, err = k.kv.Update(k.ctx, key, value, revision)
}
if errors.Is(err, jetstream.ErrKeyExists) || (err != nil && isWrongSequence(err)) {
return errSwapped
}
return err
}
func isWrongSequence(err error) bool {
var ae *jetstream.APIError
return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
}
// pipelineEnds follows the runner's ends. A nil *pipelineEnds is the trunk-only want, kept until the first registrations
// (design 47 G §95, §353).
type pipelineEnds struct {
st endStore
wake func()
logf func(string, ...any)
mu sync.Mutex
// firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design
// 47 G §14).
firsts map[string]string
}
func newPipelineEnds(st endStore, wake func(), logf func(string, ...any)) *pipelineEnds {
return &pipelineEnds{st: st, wake: wake, logf: logf, firsts: map[string]string{}}
}
// take keeps one pipeline's end if it is newer than the one kept for its module and branch, by compare-and-set.
func (e *pipelineEnds) take(p mpr.PipelineEnded, passed bool) error {
key := endKey(p.Module, p.Branch)
for {
raw, rev, err := e.st.get(key)
if err != nil {
return err
}
if raw != nil {
var kept End
if err := json.Unmarshal(raw, &kept); err == nil && !kept.newer(p.Source, p.EndCount) {
return nil // an older end, or this one again
}
}
end := End{Source: p.Source, EndCount: p.EndCount, Fp: p.Fp, Commit: p.Commit, Passed: passed, Why: p.Why}
value, err := json.Marshal(end)
if err != nil {
return err
}
if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) {
continue
} else if err != nil {
return err
}
e.wake()
return nil
}
}
// markWritten says the wants of module on branch were written from the end it keeps, if that is still the end of fp.
func (e *pipelineEnds) markWritten(module, branch, fp string) {
key := endKey(module, branch)
for {
raw, rev, err := e.st.get(key)
var kept End
if err != nil || raw == nil || json.Unmarshal(raw, &kept) != nil || kept.Fp != fp || kept.Written {
return
}
kept.Written = true
value, _ := json.Marshal(kept)
if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) {
continue
} else if err != nil {
e.logf("the end of %s on %s is not marked written: %v", module, branch, err)
}
return
}
}
// passedAt says whether the newest end kept for module on branch passed at fp.
func (e *pipelineEnds) passedAt(module, branch, fp string) bool {
raw, _, err := e.st.get(endKey(module, branch))
var kept End
return err == nil && raw != nil && json.Unmarshal(raw, &kept) == nil && kept.Passed && kept.Fp == fp
}
// branchOf is the watched branch a placement names, else the first of the module's registration, else the trunk's.
func (e *pipelineEnds) branchOf(module, placed string) string {
if placed != "" {
return placed
}
e.mu.Lock()
defer e.mu.Unlock()
if b := e.firsts[module]; b != "" {
return b
}
return trunkBranch
}
// hearPipeline takes a pipeline-ended. A pipeline of a group writes nothing on its own end, and a run on request delivers
// nothing (design 47 G §29, §64).
func (e *pipelineEnds) hearPipeline(m jetstream.Msg) {
p, err := mprc.PipelineEnded(m.Data())
switch {
case err != nil:
e.logf("%s is not taken: %v", m.Subject(), err)
case p.Group != "" || p.Run != "":
default:
if err := e.take(p, requiredPassed(p)); err != nil {
e.logf("%s: %v; it is heard again", m.Subject(), err)
_ = m.Nak()
return
}
}
_ = m.Ack()
}
// hearGroup takes a group-ended: each member's end, passed only if the group passed.
func (e *pipelineEnds) hearGroup(m jetstream.Msg) {
g, err := mprc.GroupEnded(m.Data())
if err != nil {
e.logf("%s is not taken: %v", m.Subject(), err)
_ = m.Ack()
return
}
for _, p := range g.Pipelines {
if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil {
e.logf("%s: %v; it is heard again", m.Subject(), err)
_ = m.Nak()
return
}
}
_ = m.Ack()
}
// hearEnds registers the two events on the assigner's hub.
func hearEnds(hub *process.EventHub, e *pipelineEnds) error {
if err := hub.Handle(pipelineEndedSubject, e.hearPipeline); err != nil {
return fmt.Errorf("%s: %w", pipelineEndedSubject, err)
}
if err := hub.Handle(groupEndedSubject, e.hearGroup); err != nil {
return fmt.Errorf("%s: %w", groupEndedSubject, err)
}
return nil
}
// writtenTools is the tool assigner_written: every key of the state written, with its end.
func writtenTools(st store) []stdio.Tool {
return []stdio.Tool{{Name: writtenTool, Description: "the pipeline ends the assigner's wants follow: per module and watched branch, the source revision and end count of the newest end heard, whether it passed and whether the wants were written from it.",
Run: func(map[string]any) (any, error) {
ks, err := st.keys(writtenState)
if err != nil {
return nil, err
}
sort.Strings(ks)
out := map[string]End{}
for _, k := range ks {
raw, _, ok, err := st.get(writtenState, k)
if err != nil || !ok {
continue
}
var e End
if json.Unmarshal(raw, &e) == nil {
out[k] = e
}
}
return out, nil
}}}
}
// branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one.
func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string {
if e == nil {
return nil
}
seen := map[string]bool{}
var out []string
for _, b := range placed[module] {
b = e.branchOf(module, b)
if !seen[b] {
seen[b] = true
out = append(out, b)
}
}
if len(out) == 0 {
out = append(out, e.branchOf(module, ""))
}
sort.Strings(out)
return out
}
// mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already
// (a node placed later, a module taken away).
func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool {
if e == nil {
return true
}
for _, w := range wanted {
if w.Fp == fp {
return true
}
}
for _, b := range branches {
if !e.passedAt(module, b, fp) {
return false
}
}
return true
}
// wrote marks the end the wants of module were written from.
func (e *pipelineEnds) wrote(module, fp string, branches []string) {
if e == nil {
return
}
for _, b := range branches {
e.markWritten(module, b, fp)
}
}
@@ -0,0 +1,174 @@
package main
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"mesh-delivery-assigner/internal/buckets"
)
// memEnds is endStore over a map, with revisions.
type memEnds struct {
vals map[string][]byte
revs map[string]uint64
}
func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} }
func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil }
func (m *memEnds) swap(key string, v []byte, rev uint64) error {
if m.revs[key] != rev {
return errSwapped
}
m.vals[key], m.revs[key] = v, rev+1
return nil
}
func pipe(module, branch string, source, count int64, fp, outcome string, steps ...mpr.StepResult) mpr.PipelineEnded {
return mpr.PipelineEnded{Module: module, Branch: branch, Source: source, EndCount: count, Fp: fp, Outcome: outcome, Steps: steps,
Pipeline: branch + "." + module, Commit: "c"}
}
func step(status, outcome string) mpr.StepResult {
return mpr.StepResult{Step: "unit", Status: status, Outcome: outcome}
}
// Design 47 G §29, §35, §325, §352 (tests 66, 67, 89): a want follows a passed pipeline, by compare-and-set per module and branch.
func TestAWantFollowsAPassedPipeline(t *testing.T) {
const fp1, fp2 = "fp1", "fp2"
pass, fail := mpr.PipelineEndedOutcomePassed, mpr.PipelineEndedOutcomeFailed
req, opt := mpr.StepResultStatusRequired, mpr.StepResultStatusOptional
for _, c := range []struct {
name string
ends []mpr.PipelineEnded
group *mpr.GroupEnded
module string
branch string
fp string
may bool
}{
{"a failed required step writes no want", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed"))}, nil, "web", "main", fp1, false},
{"a pipeline passed writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true},
{"a failed optional step holds nothing", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"), step(opt, "failed"))}, nil, "web", "main", fp1, true},
{"a pipeline passed at another fingerprint", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp2, false},
{"an older revision ending after a newer one writes nothing", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp1, false},
{"another branch writes its own", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "beta", 5, 1, fp1, pass)}, nil, "web", "beta", fp1, true},
{"a later end of the same revision writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed")), pipe("web", "main", 5, 2, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true},
{"the same end again after a later one keeps the later", []mpr.PipelineEnded{pipe("web", "main", 5, 2, fp1, pass), pipe("web", "main", 5, 1, fp1, fail)}, nil, "web", "main", fp1, true},
{"a member's own end writes nothing", []mpr.PipelineEnded{func() mpr.PipelineEnded { p := pipe("web", "main", 5, 1, fp1, pass); p.Group = "g"; return p }()}, nil, "web", "main", fp1, false},
{"a group ended passed writes its members", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomePassed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, true},
{"a group ended failed writes none", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomeFailed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, false},
} {
woken := 0
e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf)
for _, p := range c.ends {
if p.Group != "" {
continue // heard and ignored, as hearPipeline does
}
if err := e.take(p, requiredPassed(p)); err != nil {
t.Fatal(err)
}
}
if g := c.group; g != nil {
for _, p := range g.Pipelines {
if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil {
t.Fatal(err)
}
}
}
if got := e.mayWant(c.module, c.fp, []string{c.branch}, nil); got != c.may {
t.Errorf("%s: may want = %v, want %v", c.name, got, c.may)
}
}
// A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it).
e := newPipelineEnds(newMemEnds(), func() {}, t.Logf)
if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) {
t.Error("a want naming the fingerprint was held")
}
// markWritten keeps the end and says it was written from; a newer end replaces it unwritten.
st := newMemEnds()
e = newPipelineEnds(st, func() {}, t.Logf)
_ = e.take(pipe("web", "main", 5, 1, fp1, pass), true)
e.wrote("web", fp1, []string{"main"})
var end End
_ = json.Unmarshal(st.vals[endKey("web", "main")], &end)
if !end.Written || end.Source != 5 {
t.Errorf("end = %+v, want written at source 5", end)
}
_ = e.take(pipe("web", "main", 6, 1, fp2, pass), true)
end = End{}
_ = json.Unmarshal(st.vals[endKey("web", "main")], &end)
if end.Written || end.Source != 6 {
t.Errorf("a newer end = %+v, want unwritten at source 6", end)
}
}
// Design 47 G §29: the wants of a pass follow a pipeline: with the follower on, a module whose fingerprint no pipeline passed at
// gets no want and one a pipeline passed at gets it; with it off (nil, the trunk-only want) the pass is as before.
func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
for _, c := range []struct {
name string
follow bool
passed bool
want bool
}{{"the trunk-only want", false, false, true}, {"no pipeline passed", true, false, false}, {"a pipeline passed", true, true, true}} {
wantsKV := &memKV{values: map[string][]byte{}}
p := newPasser(kvSet{buckets.Wants: wantsKV, buckets.Asks: &memKV{values: map[string][]byte{}}}, time.Now())
p.say = func(oc.Condition) error { return nil }
mdigest := strings.Repeat("e", 64)
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
p.w.put(buckets.Modules, "web", mod)
p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}}
bkey, _ := builds.BuilderBuildKey("web", fp)
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest})
p.w.put(buckets.Builds, bkey, built)
gkey, _ := grantsstate.PartGrantKey("web", fp)
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
p.w.put(buckets.Grants, gkey, grant)
digest := "sha256:" + strings.Repeat("c", 64)
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
p.w.put(buckets.Have, "g14.web", have)
composes(t, p.w, "g14", digest)
placeOn(t, p.w, "g14", "web")
on, _ := presenceValue(true)
p.w.put(buckets.Presence, "g14", on)
if c.follow {
p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf)
if c.passed {
_ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true)
}
}
p.wants(context.Background())
key, _ := mwc.WantKey("g14", "web")
if _, got := wantsKV.values[key]; got != c.want {
t.Errorf("%s: a want written = %v, want %v", c.name, got, c.want)
}
if c.follow && c.passed {
raw, _, _ := p.ends.st.get(endKey("web", trunkBranch))
var end End
_ = json.Unmarshal(raw, &end)
if !end.Written {
t.Errorf("%s: the end was not marked written", c.name)
}
}
}
}
@@ -404,6 +404,7 @@ func serveTools() error {
tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...)
// What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it.
tools = append(tools, enginereads.Tool(engineReadsTool, "the assigner", enginereads.KeptState(stdio.State(engineReadsState)), false, time.Now))
tools = append(tools, writtenTools(runtimeStore{})...)
tools = append(tools, askAgainTools(runtimeStore{}, stdio.Ask, time.Now)...)
return stdio.Serve("", tools)
}
@@ -315,6 +315,9 @@ func (s *wantsTools) tools() []stdio.Tool {
}
var out []stdio.Tool
for _, v := range meshwants.Verbs() {
if run[v] == nil {
continue // a verb of the seat this module does not serve yet: revision, entry, difference (AS11), input (AS5)
}
out = append(out, stdio.Tool{Name: meshwants.Seat + "." + v, Run: run[v]})
}
return out
@@ -86,7 +86,7 @@ func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at t
if bucket == plc.AssignmentsBucket {
_, _ = w.assignmentKeys.Put(key, value, revision, false) // put took it already; this is its revision
}
if bucket == buckets.Source && key == mdc.SourceMainKey {
if bucket == buckets.Source && key == mdc.SourceKey(trunkBranch) {
if s, err := mdc.DecodeSource(value); err == nil && s.Group != "" {
w.sourceGroups[revision] = s.Group
}
@@ -575,3 +575,20 @@ func (w *world) unplacedNodes() []string {
sort.Strings(out)
return out
}
// trunkBranch is the branch of the trunk-only want, kept until the first registrations (design 47 G §95, §353).
const trunkBranch = "main"
// placedBranches is, by module and node, the watched branch each placement names ("" when it names none).
func (w *world) placedBranches() map[string]map[string]string {
out := map[string]map[string]string{}
for node, a := range w.assignments() {
for _, m := range a.Modules {
if out[m.Module] == nil {
out[m.Module] = map[string]string{}
}
out[m.Module][node] = m.Branch
}
}
return out
}
@@ -9,6 +9,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"mesh-delivery-assigner/internal/puts"
"reflect"
"slices"
"strings"
"testing"
@@ -431,7 +432,7 @@ func TestEachRuleHoldsTheWantBack(t *testing.T) {
before["g14"] = mwc.Want{Fp: fNew, Composed: dig(fNew)}
}
out := Wants(m, n, before, nil)
if got := out.Wants["g14"]; got != c.want {
if got := out.Wants["g14"]; !reflect.DeepEqual(got, c.want) {
t.Errorf("%s: g14's want = %+v, want %+v", c.name, got, c.want)
}
if got := out.Rules["g14"].Rule; got != c.rule {
@@ -65,7 +65,7 @@ func TestManifest(t *testing.T) {
}
if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
"assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1,
"engine-reads": 1}; !reflect.DeepEqual(states, want) {
"engine-reads": 1, "written": 5}; !reflect.DeepEqual(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
@@ -96,7 +96,7 @@ func TestManifest(t *testing.T) {
// which lists and asks again what each node's engine reads.
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants",
"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies",
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again"}) {
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) {
t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools)
}
if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs {
@@ -123,7 +123,7 @@ func TestManifest(t *testing.T) {
}
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "node-engine.manifest-reads-changed"}) {
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) {
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes)
}
if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
+6 -2
View File
@@ -45,6 +45,8 @@
"mesh-delivery-indexer.views.manifests",
"mesh-delivery-indexer.views.modules",
"mesh-delivery-teller.ask-ended",
"mesh-pipeline-runner.pipeline-ended",
"mesh-pipeline-runner.group-ended",
"node-engine.manifest-reads-changed"
],
"invokes": [
@@ -90,7 +92,8 @@
"assigner_engine_reads",
"assigner_seat_list",
"assigner_seat_list_again",
"assigner_ask_again"
"assigner_ask_again",
"assigner_written"
],
"state": [
{ "name": "assignments", "history": 64 },
@@ -108,7 +111,8 @@
{ "name": "seat-holders", "history": 5 },
{ "name": "view-marks", "history": 1 },
{ "name": "first-views", "history": 1 },
{ "name": "engine-reads", "history": 1 }
{ "name": "engine-reads", "history": 1 },
{ "name": "written", "history": 5 }
],
"own-secrets": {
"broker": "${dir:state}/broker"
@@ -409,7 +409,16 @@ func main() {
}
return e.Value, true, nil
}}
rs := stdio.State(versions.ReservedState)
rr := versions.Reader{Keys: rs.Keys, Get: func(key string) (json.RawMessage, bool, error) {
e, err := rs.Get(key)
if err != nil || e == nil {
return nil, false, err
}
return e.Value, true, nil
}}
tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(builds.BuilderLog).Since, time.Now, vr, tags, tr, fr, ofr)
tools = append(tools, reservedTools(rr, buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
// The builder's expiring keys (state bounds): what each bounds and when it ends.
tools = append(tools, stdio.Tool{Name: versions.BoundsTool, Description: versions.BoundsDescription,
Input: map[string]any{"type": "object", "properties": map[string]any{}},
@@ -0,0 +1,62 @@
package main
import (
"errors"
"fmt"
"strings"
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// reservedTools are builder_reserved, builder_reserved_drop and the seat mesh-builds' verb reserved (design 47 G section
// 369), served as tools of the bundle. The drop is a hand act: it is recorded in the hand-act log first, then written to
// reasks for the builder, which owns the reservations, to answer.
func reservedTools(rr versions.Reader, ask queueAsker, now func() time.Time, acts meshhandactsclient.Recorder) []stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
return []stdio.Tool{
{Name: versions.ReservedTool, Description: versions.ReservedDescription,
Input: map[string]any{"type": "object", "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "only this module's reservations"}}},
Run: func(args map[string]any) (any, error) { return versions.AnswerReserved(rr, str(args, "module")) }},
{Name: versions.ReservedVerb, Description: "The seat mesh-builds' verb reserved: a module's reserved version with each artifact's digest once built " +
"(design 47 G section 369), for a registry holder deciding whether to publish. Read-only.",
Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "the module"},
"version": map[string]any{"type": "string", "description": "the version, when asked for a particular one"}}},
Run: func(args map[string]any) (any, error) {
return versions.AnswerReservedVerb(rr, str(args, "module"), str(args, "version"))
}},
{Name: versions.ReservedDropTool, Description: versions.ReservedDropDescription,
Input: map[string]any{"type": "object", "required": []string{"module", "series", "why"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "the module"},
"series": map[string]any{"type": "string", "description": "its MAJOR.MINOR"},
"why": map[string]any{"type": "string", "description": "why, in plain words"}}},
Run: func(args map[string]any) (any, error) {
key, value, err := build.NewReservedDrop(str(args, "module"), str(args, "series"), str(args, "why"), stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
if _, err := acts.Record(meshhandactsclient.Request{What: fmt.Sprintf("dropped the reservation of %s %s by hand", str(args, "module"), str(args, "series")),
Why: str(args, "why"), Cause: versions.ReservedDropTool, Verb: "reserved-drop"}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was dropped: %w", err)
}
raw, err := ask(key, value)
if err != nil {
return nil, err
}
a, err := build.DecodeReAsk(raw)
if err != nil {
return nil, err
}
if a.Refused {
return nil, errors.New(a.Answer)
}
return map[string]any{"key": key, "answer": a.Answer}, nil
}},
}
}
@@ -623,7 +623,7 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
}
// Running-testing note (review round 1): two claims never get one number on the bus's own compare-and-set, as
// the builder's state versions keeps them: eight claims of one name at once, eight versions.
store := versions.Versions{Store: awaitedVersions{process.AwaitBucket(b.ctx, b.js, nil, versions.Bucket, "", t.Logf)}}
store := versions.Versions{Store: awaitedVersions{b: process.AwaitBucket(b.ctx, b.js, nil, versions.Bucket, "", t.Logf)}}
numbers := make(chan string, 8)
var claims sync.WaitGroup
for i := range 8 {
@@ -163,7 +163,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
floorWaits.Keep(kvKeeper{ctx: ctx, kv: kv}, heldFloorWaits(ctx, kv, logger.Printf))
})
// A package is numbered under its name in its ecosystem, read from the indexer's packages state (design 32).
numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits,
numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{b: versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits,
Names: versions.NamesIn(indexerPackages{packages}),
Said: func(nb builds.VersionNumbered) {
if err := emitNumbered(context.WithoutCancel(ctx), js, cred, nb); err != nil {
@@ -202,10 +202,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
boundsKV := process.AwaitBucket(ctx, js, conn, versions.TagAskBoundsBucket, "an unanswered tag ask is not narrated", logger.Printf)
bounds := &tagBounds{asks: tagAsksKV.KV, bounds: boundsKV.KV, bound: tagAskBound, waits: floorWaits.Waiting, logf: logger.Printf, now: time.Now,
log: func(e builds.BuildLogEntry) { b.Log(e) }}
// The versions a build reserves at its start (design 47 G §369): its state reserved, and the catalog's versions-filled
// and version-released heard; nothing is reserved before the fill.
reservedKV := process.AwaitBucket(ctx, js, conn, versions.ReservedBucket, "no version is reserved: builds are made unreserved", logger.Printf)
reserved := newReservations(awaitedReserved{awaitedVersions{b: reservedKV, bucket: versions.ReservedBucket}}, calls.Call, pubBoundsKV.KV, found, b.Reconsider, logger.Printf)
b.Reserved = reserved
// A state bound later numbers the builds looked at without it.
versionsKV.Ready(func(jetstream.KeyValue) { b.Reconsider(ctx) })
// The versions' hand acts (builder_publish_again, builder_floor_again) come through reasks, as builder_build_again.
acts := versions.Acts{Numberer: numberer, Done: doneVia(b), Reconsider: b.Reconsider, Remove: removeVia(calls.Call),
acts := versions.Acts{Numberer: numberer, DropReserved: reserved.dropped, Done: doneVia(b), Reconsider: b.Reconsider, Remove: removeVia(calls.Call),
BuildAgain: func(ctx context.Context, module, fp, why, by string) (string, bool) {
r := b.BuildAgain(ctx, module, fp, why, by, buildAgainBound)
if r.Refused != "" {
@@ -349,9 +354,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// condition keeper's condition-cleared.
floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf}
// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs}
seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs, res: reserved}
events := func(m jetstream.Msg) {
if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || reserved.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
return
}
heardGone(ctx, b, m, logger.Printf)
@@ -407,6 +412,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
})
pubBoundsKV.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) {
if reserved.ended(ctx, e) {
return
}
if _, ok := versions.UnputBoundOf(e.Key()); ok {
unput.expired(ctx, e)
return
@@ -53,7 +53,7 @@ func reAsked(ctx context.Context, b askAgainer, acts versions.Acts, kv jetstream
announcedAgain(ctx, b, kv, e, r, logf)
return
}
if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop {
if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop || r.Mode == build.ModeReservedDrop {
// A version's hand act (design 51 *Tools of the builder*) may build again, so it is answered beside the watch.
go versionActed(ctx, acts, kv, e, r, logf)
return
@@ -163,6 +163,12 @@ func versionActed(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue
var ag versions.Again
ag, r.Answer, err = acts.PublishAgain(ctx, r.Name, r.FP, r.Why, r.By, r.At)
r.Do, r.Version = ag.Do, ag.Entry.Version
} else if r.Mode == build.ModeReservedDrop {
if acts.DropReserved == nil {
err = errors.New("this builder keeps no reservations")
} else {
r.Answer, err = acts.DropReserved(ctx, r.Name, r.FP, r.By, r.Why, r.At)
}
} else if r.Mode == build.ModeOldFormDrop {
if acts.DropOldForm == nil {
err = errors.New("this builder keeps no old-form window")
@@ -0,0 +1,208 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync/atomic"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// The reservations (design 47 G §369): the builder's state reserved, the catalog's versions it numbers from, and the
// catalog's version-released and versions-filled it hears.
// awaitedReserved is the state reserved as versions.ReserveStore: bound once the bus holds it.
type awaitedReserved struct{ awaitedVersions }
func (s awaitedReserved) Delete(ctx context.Context, key string) error {
kv, err := s.kv()
if err != nil {
return err
}
if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
return err
}
return nil
}
// reservations is the builder's reserving: the Reserver, the fill the catalog said, the bounds of a built reservation's
// publish and the findings.
type reservations struct {
r *versions.Reserver
filled atomic.Bool
bounds func() jetstream.KeyValue
found *findings
bound time.Duration
wake func(context.Context)
logf func(string, ...any)
}
// newReservations makes the reserving over store; the released versions of a module are asked of the seat mesh-catalog's
// verb versions through call (design 47 G §348), the only source numbering reads them from.
func newReservations(store versions.ReserveStore, call toolCaller, bounds func() jetstream.KeyValue, found *findings, wake func(context.Context), logf func(string, ...any)) *reservations {
rv := &reservations{bounds: bounds, found: found, bound: versions.UnreleasedBound, wake: wake, logf: logf}
rv.r = &versions.Reserver{Store: store, Filled: rv.filled.Load, ReleasedOf: func(_ context.Context, module string) ([]string, error) {
raw, err := call("seat:"+meshcatalog.Seat+".versions", meshcatalog.VersionsInput{Module: module})
if err != nil {
return nil, err
}
var a meshcatalog.VersionsAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return nil, fmt.Errorf("the catalog's versions of %s answered what is no list of versions: %w", module, err)
}
var out []string
for _, v := range a.Versions {
out = append(out, v.Version)
}
return out, nil
}}
return rv
}
var _ build.Reservations = (*reservations)(nil)
func (rv *reservations) Filled() bool { return rv.filled.Load() }
func (rv *reservations) Reserve(ctx context.Context, module, series string, artifacts []string) (versions.ReservedVersion, error) {
return rv.r.Reserve(ctx, module, series, artifacts)
}
// Built records the digests; once every artifact has one, the publish's bound starts (a key of the state bounds with its
// own time to live): its end with the reservation standing is version-unreleased.
func (rv *reservations) Built(ctx context.Context, module, version string, digests map[string]string) error {
for ref, d := range digests {
if _, err := rv.r.Built(ctx, module, version, ref, d); err != nil {
return err
}
}
cur, found, err := rv.r.Of(ctx, module, version)
if err != nil || !found || cur.BuiltAt.IsZero() {
return err
}
rv.arm(ctx, cur)
return nil
}
func (rv *reservations) arm(ctx context.Context, v versions.ReservedVersion) {
kv := rv.bounds()
var err error
if kv == nil {
err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket)
} else if _, err = kv.Create(ctx, versions.ReservedBoundKey(v.Module, v.Series), versions.EncodeBound("the release of "+v.Module+" "+v.Version, time.Now().Add(rv.bound)), jetstream.KeyTTL(rv.bound)); errors.Is(err, jetstream.ErrKeyExists) {
err = nil
}
if err != nil {
rv.logf("the release bound of %s %s is not written, so it is raised unreleased now: %v", v.Module, v.Version, err)
rv.unreleased(ctx, v, fmt.Sprintf("its bound could not be written (%v)", err))
}
}
// unreleased raises version-unreleased about the module, naming the version and the artifacts missing.
func (rv *reservations) unreleased(ctx context.Context, v versions.ReservedVersion, why string) {
missing := v.Missing()
what := "an artifact is unpublished"
if len(missing) > 0 {
what = strings.Join(missing, ", ") + " unpublished"
}
rv.found.raise(ctx, finding{kind: versions.VersionUnreleased, scope: meshconditionsclient.ScopeModule, id: v.Module,
summary: fmt.Sprintf("%s %s is built and not released: %s (%s)", v.Module, v.Version, what, why),
headline: "A built version is not released",
explanation: "The builder reserved " + v.Module + " " + v.Version + " and its build ended, and the catalog has not said version-released: " + what + ".",
resolved: "The catalog says version-released for " + v.Module + " " + v.Version + ".",
needs: "builder_reserved shows the reservation; builder_publish_again publishes the missing artifacts under it; builder_reserved_drop drops it."})
}
// ended is told each change of the state bounds: a reservation's bound whose time ended raises version-unreleased while
// the reservation stands built.
func (rv *reservations) ended(ctx context.Context, e jetstream.KeyValueEntry) bool {
key, ok := versions.ReservedBoundOf(e.Key())
if !ok {
return false
}
if e.Operation() != jetstream.KeyValuePurge {
return true
}
v, found, err := rv.r.At(ctx, key)
if err != nil || !found || v.BuiltAt.IsZero() {
return true
}
rv.unreleased(ctx, v, fmt.Sprintf("nothing within %s of its build", rv.bound))
return true
}
// dropBound ends a reservation's bound.
func (rv *reservations) dropBound(ctx context.Context, key string) {
if kv := rv.bounds(); kv != nil {
if err := kv.Delete(ctx, versions.ReservedBoundFor(key)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
rv.logf("the release bound of %s is not ended: %v", key, err)
}
}
}
// heard takes the catalog's version-released and versions-filled in, and a registry seat's published (published): a
// message that is none of them is left to the others. A state that cannot take one leaves it for the bus to give again.
func (rv *reservations) heard(ctx context.Context, m jetstream.Msg) bool {
switch m.Subject() {
case meshcatalog.SubjectVersionsFilled:
rv.filled.Store(true)
rv.logf("the catalog said versions-filled: versions are reserved from now on")
if rv.wake != nil {
rv.wake(ctx)
}
case meshcatalog.SubjectVersionReleased:
var ev meshcatalog.VersionReleasedEvent
if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Module == "" || ev.Version == "" {
rv.logf("%s holds what is no version-released, left: %v", m.Subject(), err)
break
}
gone, err := rv.r.Released(ctx, ev.Module, ev.Version)
if err != nil {
rv.logf("%v; heard again", err)
if nakErr := m.Nak(); nakErr != nil {
rv.logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
}
return true
}
for _, k := range gone {
rv.dropBound(ctx, k)
}
rv.found.clear(ctx, versions.VersionUnreleased, meshconditionsclient.ScopeModule, ev.Module)
default:
return false
}
if err := m.Ack(); err != nil {
rv.logf("acknowledging %s: %v", m.Subject(), err)
}
return true
}
// published marks the artifacts of the reservation a registry seat's published names (its version and digest).
func (rv *reservations) published(ctx context.Context, o versions.Outcome) {
if _, err := rv.r.Published(ctx, o.Version, o.Digest); err != nil {
rv.logf("a published of %s@%s is not recorded in the reservations: %v", o.Name, o.Version, err)
}
}
// dropped is builder_reserved_drop: the reservation is deleted and its bound ended.
func (rv *reservations) dropped(ctx context.Context, module, series, by, why string, at time.Time) (string, error) {
ok, err := rv.r.Drop(ctx, module, series)
if err != nil {
return "", err
}
if !ok {
return "", fmt.Errorf("%s holds no reservation of %s", module, series)
}
rv.dropBound(ctx, versions.ReservedKey(module, series))
rv.found.clear(ctx, versions.VersionUnreleased, meshconditionsclient.ScopeModule, module)
return fmt.Sprintf("the reservation of %s %s is dropped (%s, %s)", module, series, by, why), nil
}
@@ -0,0 +1,131 @@
package main
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-builder/internal/versions"
)
// memReserve is the state reserved in memory.
type memReserve struct {
vals map[string][]byte
revs map[string]uint64
}
func (m *memReserve) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
v, ok := m.vals[key]
return v, m.revs[key], ok, nil
}
func (m *memReserve) Create(_ context.Context, key string, v []byte) (uint64, error) {
if _, ok := m.vals[key]; ok {
return 0, versions.ErrConflict
}
m.vals[key], m.revs[key] = v, 1
return 1, nil
}
func (m *memReserve) Update(_ context.Context, key string, v []byte, rev uint64) (uint64, error) {
if m.revs[key] != rev {
return 0, versions.ErrConflict
}
m.vals[key], m.revs[key] = v, rev+1
return rev + 1, nil
}
func (m *memReserve) KeysOf(_ context.Context, filter string) ([]string, error) {
var out []string
for k := range m.vals {
if filter == versions.AllFilter || strings.HasPrefix(k, strings.TrimSuffix(filter, "*")) {
out = append(out, k)
}
}
return out, nil
}
func (m *memReserve) Delete(_ context.Context, key string) error {
delete(m.vals, key)
return nil
}
// boundsKV keeps the keys created and deleted in the state bounds.
type boundsKV struct {
jetstream.KeyValue
keys map[string]bool
}
func (b *boundsKV) Create(_ context.Context, key string, _ []byte, _ ...jetstream.KVCreateOpt) (uint64, error) {
b.keys[key] = true
return 1, nil
}
func (b *boundsKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
delete(b.keys, key)
return nil
}
// TestTheCatalogsWordsMoveTheReservations: design 47 G §369, §372 (test 102) — versions-filled opens the reserving and wakes
// the builds that wait; a built reservation arms its bound and its end raises version-unreleased naming the artifacts
// missing; the catalog's version-released deletes the key and the bound and clears the finding; builder_reserved_drop drops one.
func TestTheCatalogsWordsMoveTheReservations(t *testing.T) {
ctx := context.Background()
st := &memFindings{vals: map[string][]byte{}}
found := &findings{store: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }}
bounds := &boundsKV{keys: map[string]bool{}}
woke := 0
call := func(addr string, body any) (json.RawMessage, error) {
if addr != "seat:mesh-catalog.versions" {
t.Errorf("the released versions were asked of %q", addr)
}
return json.Marshal(meshcatalog.VersionsAnswer{Module: "m", Versions: []meshcatalog.VersionEntry{{Module: "m", Version: "1.4.2"}}})
}
rv := newReservations(&memReserve{vals: map[string][]byte{}, revs: map[string]uint64{}}, call, func() jetstream.KeyValue { return bounds }, found,
func(context.Context) { woke++ }, t.Logf)
if _, err := rv.Reserve(ctx, "m", "1.4", []string{"m:a"}); err == nil || rv.Filled() {
t.Fatalf("a version was reserved before versions-filled (%v)", err)
}
filled := &wakeMsg{subject: meshcatalog.SubjectVersionsFilled, data: []byte(`{"v":1}`)}
if !rv.heard(ctx, filled) || !filled.acked || !rv.Filled() || woke != 1 {
t.Fatalf("versions-filled: acked %v, filled %v, woke %d", filled.acked, rv.Filled(), woke)
}
res, err := rv.Reserve(ctx, "m", "1.4", []string{"m:a", "m:b"})
if err != nil || res.Version != "1.4.3" {
t.Fatalf("reserved %q (%v), want 1.4.3 after the catalog's 1.4.2", res.Version, err)
}
if err := rv.Built(ctx, "m", "1.4.3", map[string]string{"m:a": "sha256:a", "m:b": "sha256:b"}); err != nil {
t.Fatal(err)
}
boundKey := versions.ReservedBoundKey("m", "1.4")
if !bounds.keys[boundKey] {
t.Fatalf("the publish's bound is not armed: %v", bounds.keys)
}
rv.published(ctx, versions.Outcome{Name: "m-a", Version: "1.4.3", Digest: "sha256:a"})
if !rv.ended(ctx, purged{key: boundKey}) {
t.Fatal("a reservation's bound was not taken in")
}
key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, "m", versions.VersionUnreleased)
got, err := meshconditionsclient.DecodeFinding(st.vals[key])
if err != nil || !strings.Contains(got.Summary, "m:b unpublished") || strings.Contains(got.Summary, "m:a unpublished") || !strings.Contains(got.Summary, "1.4.3") {
t.Fatalf("finding = %+v (%v), want version-unreleased naming m:b of 1.4.3", got, err)
}
released := &wakeMsg{subject: meshcatalog.SubjectVersionReleased, data: []byte(`{"v":1,"module":"m","version":"1.4.3","source":1,"artifacts":[]}`)}
if !rv.heard(ctx, released) || !released.acked {
t.Fatalf("version-released was not taken in (acked %v)", released.acked)
}
if all, _ := rv.r.All(ctx); len(all) != 0 || bounds.keys[boundKey] || len(st.vals) != 0 {
t.Errorf("after version-released: reservations %v, bounds %v, findings %v; want none", all, bounds.keys, st.vals)
}
if _, err := rv.Reserve(ctx, "m", "1.4", []string{"m:a"}); err != nil {
t.Fatal(err)
}
if answer, err := rv.dropped(ctx, "m", "1.4", "op", "never released", time.Unix(0, 0)); err != nil || !strings.Contains(answer, "dropped") {
t.Errorf("builder_reserved_drop: %q (%v)", answer, err)
}
if _, err := rv.dropped(ctx, "m", "1.4", "op", "again", time.Unix(0, 0)); err == nil {
t.Error("a reservation that is gone was dropped again")
}
}
@@ -50,6 +50,8 @@ type seatHearer struct {
logf func(string, ...any)
// pub raises and clears the findings of a publish that ended badly; nil raises none.
pub *publishes
// res marks a reservation's artifacts published; nil marks none.
res *reservations
}
// heard takes a message of the events consumer in when it is a registry seat's outcome, and answers whether it was.
@@ -90,6 +92,9 @@ func (h *seatHearer) heard(ctx context.Context, m jetstream.Msg) bool {
if h.pub != nil {
h.pub.heard(ctx, se.seat, se.event, o)
}
if h.res != nil && se.event == versions.HeardPublished {
h.res.published(ctx, o)
}
h.ack(m)
if len(wake) > 0 {
h.logf("the floor of %s is asked again: %s said %s", strings.Join(wake, ", "), se.seat, se.event)
@@ -29,7 +29,11 @@ const (
// awaitedVersions is the state versions as versions.Store: bound once the bus holds it (the grants part makes it, issue
// 1790); until then every use is an error, so a trunk build is built unnumbered and said (build.Builder.number).
type awaitedVersions struct{ b *process.AwaitedBucket }
type awaitedVersions struct {
b *process.AwaitedBucket
// bucket names the state in the error of one not on the bus yet; empty is the versions'.
bucket string
}
var errVersionsMissing = errors.New("the state " + versions.Bucket + " is not on the bus yet")
@@ -37,6 +41,9 @@ func (s awaitedVersions) kv() (jetstream.KeyValue, error) {
if kv := s.b.KV(); kv != nil {
return kv, nil
}
if s.bucket != "" {
return nil, errors.New("the state " + s.bucket + " is not on the bus yet")
}
return nil, errVersionsMissing
}
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"strconv"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
@@ -140,3 +141,59 @@ func (b *Builder) PackageNamed(ctx context.Context, module string) {
b.consider(ctx, module, m)
}
}
// Reservations reserves the version of a trunk build at its start and records each artifact's digest (design 47 G §369).
// Nil reserves nothing.
type Reservations interface {
// Filled says the catalog said versions-filled: nothing is reserved before it.
Filled() bool
// Reserve reserves module's next version in series for the artifacts (their references, <module>:<artifact>).
Reserve(ctx context.Context, module, series string, artifacts []string) (versions.ReservedVersion, error)
// Built records the digests of the artifacts built under the reserved version, by reference.
Built(ctx context.Context, module, version string, digests map[string]string) error
}
// reserves says whether a look at a trunk build is one that reserves: the build has a series of the form MAJOR.MINOR
// (design 47 G §368) and artifacts, and its key is not done already.
func (b *Builder) reserves(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan) bool {
if b.Reserved == nil || len(plan.Tasks) == 0 {
return false
}
if strings.Count(plan.Series, ".") != 1 {
return false
}
return b.doneDigests(ctx, module, m.Fp2) == nil
}
// waitFill holds a build that would reserve until the catalog says versions-filled, saying wait-held naming the fill once
// (design 47 G §369, §372). It answers whether the build waits.
func (b *Builder) waitFill(module string, m meshcatalog.ModulesByModuleValue) bool {
if b.Reserved.Filled() {
return false
}
if _, said := b.fillWaits.LoadOrStore(module+"."+m.Fp2, true); !said {
line := module + " at " + m.Fp2 + " is not built yet: wait-held, the catalog has not said versions-filled, and nothing is reserved before it"
b.logf("%s", line)
if b.Log != nil {
b.Log(builds.BuildLogEntry{Event: builds.BuildSaidEvent, Module: module, FP: m.Fp2, Commit: m.Commit, Node: b.Node,
At: time.Now().UTC(), Line: line, Why: "wait-held"})
}
}
return true
}
// reserve reserves the version of the build that starts; a reservation that cannot be made is said and the build goes on
// unreserved, as an unnumbered build does.
func (b *Builder) reserve(ctx context.Context, module string, plan Plan) (versions.ReservedVersion, bool) {
refs := make([]string, 0, len(plan.Tasks))
for _, t := range plan.Tasks {
refs = append(refs, module+":"+t.Artifact)
}
res, err := b.Reserved.Reserve(ctx, module, plan.Series, refs)
if err != nil {
b.logf("%s: no version is reserved for this build: %v", module, err)
return versions.ReservedVersion{}, false
}
b.logf("%s: version %s is reserved for this build", module, res.Version)
return res, true
}
@@ -35,7 +35,11 @@ type Task struct {
}
// Plan is what a manifest asks the builder to build, by artifact name.
type Plan struct{ Tasks []Task }
type Plan struct {
Tasks []Task
// Series is the manifest's MAJOR.MINOR (design 47 G §368): the series its build reserves a version in (§369).
Series string
}
// Modules are the build modules the plan calls, sorted.
func (p Plan) Modules() []string {
@@ -89,7 +93,7 @@ func PlanOf(m catalogue.Manifest) (Plan, error) {
}
artifacts := append([]catalogue.Artifact(nil), m.Build.Artifacts...)
sort.Slice(artifacts, func(i, j int) bool { return artifacts[i].Name < artifacts[j].Name })
var p Plan
p := Plan{Series: m.Version}
for _, a := range artifacts {
switch {
case a.Kind == catalogue.ArtifactArchive:
@@ -0,0 +1,104 @@
package build
import (
"context"
"slices"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"mesh-delivery-builder/internal/versions"
)
// fakeReservations answers a fill and keeps what the builder reserved and recorded.
type fakeReservations struct {
filled bool
reserved [][]string
series []string
built map[string]string
}
func (f *fakeReservations) Filled() bool { return f.filled }
func (f *fakeReservations) Reserve(_ context.Context, module, series string, artifacts []string) (versions.ReservedVersion, error) {
f.reserved, f.series = append(f.reserved, artifacts), append(f.series, series)
return versions.ReservedVersion{Module: module, Series: series, Version: series + ".5"}, nil
}
func (f *fakeReservations) Built(_ context.Context, _, version string, digests map[string]string) error {
f.built = map[string]string{version: ""}
for ref, d := range digests {
f.built[ref] = d
}
return nil
}
// TestABuildReservesItsVersionAtItsStartOnceTheCatalogFilled: design 47 G §369, §372 (test 102) — nothing is reserved before
// versions-filled and the build waits saying wait-held naming the fill; after it the build reserves one version in the
// manifest's MAJOR.MINOR for its artifacts and records their digests; a done key reserves nothing again; a manifest whose
// version is no MAJOR.MINOR reserves nothing.
func TestABuildReservesItsVersionAtItsStartOnceTheCatalogFilled(t *testing.T) {
r := newRig(t, goBundle)
r.mf.byDigest[mdigest('a')] = catalogue.Manifest{Module: "m", Version: "0.1", Build: &catalogue.Build{Artifacts: []catalogue.Artifact{goBundle}}}
res := &fakeReservations{}
r.b.Reserved = res
var said []builds.BuildLogEntry
r.b.Log = func(e builds.BuildLogEntry) { said = append(said, e) }
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
r.b.Wait()
if _, ok := r.st.vals[r.key]; ok || len(res.reserved) != 0 {
t.Fatalf("before versions-filled the build was claimed (%v) or reserved %v", ok, res.reserved)
}
var waits []string
for _, e := range said {
if e.Event == builds.BuildSaidEvent && e.Why == "wait-held" {
waits = append(waits, e.Line)
}
}
if len(waits) != 1 || !slices.ContainsFunc(waits, func(l string) bool { return strings.Contains(l, "versions-filled") }) {
t.Fatalf("wait-held lines = %q, want one naming versions-filled", waits)
}
res.filled = true
v := r.consider(t, "modules/m")
if v.State != builds.BuildDone || len(res.reserved) != 1 || !slices.Equal(res.reserved[0], []string{"m:tool"}) || res.series[0] != "0.1" {
t.Fatalf("build %+v reserved %v in %v, want done and one reservation of m:tool in 0.1", v.State, res.reserved, res.series)
}
if res.built["m:tool"] != digest || len(res.built) != 2 {
t.Errorf("recorded %v, want the digest of m:tool under the reserved version", res.built)
}
r.consider(t, "modules/m")
if len(res.reserved) != 1 {
t.Errorf("a done key reserved again: %v", res.reserved)
}
// A manifest whose version is no MAJOR.MINOR reserves nothing and is built as before.
r2 := newRig(t, goBundle)
res2 := &fakeReservations{filled: true}
r2.b.Reserved = res2
if v := r2.consider(t, "modules/m"); v.State != builds.BuildDone || len(res2.reserved) != 0 {
t.Errorf("a manifest with no MAJOR.MINOR: state %s, reserved %v", v.State, res2.reserved)
}
}
// TestADropOfAReservationIsAReasksValueOfItsOwnKey: design 47 G §369 — builder_reserved_drop asks the builder through
// reasks, under a key of the reservation (never a build key), refusing a module with no MAJOR.MINOR or no why.
func TestADropOfAReservationIsAReasksValueOfItsOwnKey(t *testing.T) {
at := time.Unix(0, 0)
key, data, err := NewReservedDrop("m", "1.4", "never released", "op@x", at)
if err != nil || key != "reserved."+versions.ReservedKey("m", "1.4") {
t.Fatalf("key %q (%v)", key, err)
}
if r, err := DecodeReAsk(data); err != nil || r.Mode != ModeReservedDrop || r.Name != "m" || r.FP != "1.4" || r.By != "op@x" {
t.Errorf("value = %+v (%v)", r, err)
}
for name, args := range map[string][3]string{"no module": {"", "1.4", "why"}, "no series": {"m", "1", "why"}, "three numbers": {"m", "1.4.2", "why"}, "no why": {"m", "1.4", " "}} {
if _, _, err := NewReservedDrop(args[0], args[1], args[2], "op", at); err == nil {
t.Errorf("%s: a drop was made", name)
}
}
}
@@ -95,6 +95,8 @@ type Builder struct {
// Numbers numbers each trunk build at its claim and records what it was built with (design 51 *The versions*); nil
// numbers nothing.
Numbers Numbers
// Reserved reserves the version of a trunk build at its start (design 47 G §369); nil reserves nothing.
Reserved Reservations
// Tags answers a check's build's tag from the kept tag asks; nil numbers no check's build.
Tags TagOf
// Forms answers what each node's engine states it reads (ADR 0618 §2, ADR 0624 §1): present, the nodes whose engine
@@ -103,16 +105,18 @@ type Builder struct {
// its single reference, and its needs. Nil: no node states anything, and the first forms are kept.
Forms func() (present, counted map[string][]string)
once sync.Once
sem chan struct{}
wg sync.WaitGroup
mu sync.Mutex
known map[string]meshcatalog.ModulesByModuleValue // by modules key
inflight map[string]bool // by build key
again map[string]meshcatalog.ModulesByModuleValue // by build key: the newest value that arrived while the key was being built
plans map[string]planned // by manifest digest
cands map[string]meshcatalog.CandidatesByGroupValue // by candidates key (candidates.go)
via map[string]via // by build key: a candidate's fetch (candidates.go)
once sync.Once
sem chan struct{}
wg sync.WaitGroup
mu sync.Mutex
// fillWaits are the builds that said wait-held for the fill (module.fingerprint), so it is said once.
fillWaits sync.Map
known map[string]meshcatalog.ModulesByModuleValue // by modules key
inflight map[string]bool // by build key
again map[string]meshcatalog.ModulesByModuleValue // by build key: the newest value that arrived while the key was being built
plans map[string]planned // by manifest digest
cands map[string]meshcatalog.CandidatesByGroupValue // by candidates key (candidates.go)
via map[string]via // by build key: a candidate's fetch (candidates.go)
// outOfStep are, by build key, the out-of-step files a build's lanes answered, until its end entry takes them.
outOfStep map[string][]string
trunkFP map[string]bool // by build key: a trunk module value carries this fingerprint
@@ -491,6 +495,10 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
return
}
}
reserving := trunk && tag == "" && planErr == nil && !nothing && b.reserves(ctx, module, m, plan)
if reserving && b.waitFill(module, m) {
return
}
// A key built again by hand is taken over as it stands too: one left building by a look that ended without its write
// (issue 1966) is the hand tool's to finish.
cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first || byHand, ByHand: byHand, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
@@ -525,6 +533,11 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
}
}
}
var reserved versions.ReservedVersion
hasReserved := false
if reserving && cl.Build.State != builds.BuildDone {
reserved, hasReserved = b.reserve(ctx, module, plan)
}
var result builds.BuilderBuild
if nothing {
// A manifest with no artifact is built by building nothing: its key is done with no artifact, claimed and written as
@@ -580,6 +593,15 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
b.logf("%s: %v", cl.Key, err)
}
}
if result.State == builds.BuildDone && hasReserved {
digests := map[string]string{}
for _, a := range result.Artifacts {
digests[module+":"+a.Name] = a.Digest
}
if err := b.Reserved.Built(ctx, module, reserved.Version, digests); err != nil {
b.logf("%s: the digests of %s are not recorded: %v", cl.Key, reserved.Version, err)
}
}
if result.State == builds.BuildFailed {
b.logf("%s: failed: %s", cl.Key, result.Error)
}
@@ -120,6 +120,27 @@ func NewOldFormDrop(node, why, by string, at time.Time) (string, []byte, error)
return "old-form." + node, data, err
}
// ModeReservedDrop is the Mode of a reasks value of builder_reserved_drop (design 47 G section 369): a module's reservation
// of one MAJOR.MINOR dropped by hand; Name is the module and FP the MAJOR.MINOR.
const ModeReservedDrop = "reserved-drop"
// NewReservedDrop is the value builder_reserved_drop writes: the key `reserved.<module>.<MAJOR.MINOR>` (never a build key)
// and the value.
func NewReservedDrop(module, series, why, by string, at time.Time) (string, []byte, error) {
module, series = strings.TrimSpace(module), strings.TrimSpace(series)
if module == "" {
return "", nil, errors.New("module is required")
}
if strings.Count(series, ".") != 1 {
return "", nil, fmt.Errorf("series %q is not MAJOR.MINOR", series)
}
if strings.TrimSpace(why) == "" {
return "", nil, errors.New("say why the reservation is dropped (why)")
}
data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC(), Mode: ModeReservedDrop, Name: module, FP: series})
return "reserved." + versions.ReservedKey(module, series), data, err
}
// ModeBuild is the Mode of a reasks value that asks the key to be built again, not its waiting asks to be asked again.
const ModeBuild = "build"
@@ -217,6 +217,8 @@ type Acts struct {
Remove func(ctx context.Context, seat string, e Entry) (RemoveOutcome, error)
// DropOldForm drops a retired node from the old form's count by hand (builder_old_form): its answer, or why not.
DropOldForm func(ctx context.Context, node, by, why string, at time.Time) (string, error)
// DropReserved drops a module's reservation of one MAJOR.MINOR by hand (builder_reserved_drop): its answer, or why not.
DropReserved func(ctx context.Context, module, series, by, why string, at time.Time) (string, error)
}
// TagAsked answers one unanswered value of tag-asks. An untag calls remove on each of the branch's tagged versions'
@@ -0,0 +1,450 @@
package versions
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"strconv"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
)
// **The reservation is one key per module and MAJOR.MINOR** (design 47 G §369, §52): the builder keeps in its old state
// section `reserved` the version its build reserved and each artifact's digest once built, written by compare-and-set at
// the build's start. A newer build of the module takes the next after the higher of the highest released and the one
// reserved, so two builds never share a version, and a superseded build's version is never released (a gap in the
// patches). The key is deleted when the catalog says `version-released` for it, or when a version of another MAJOR.MINOR
// of the module is released (the manifest moved on). This is no list of released versions: numbering reads those only from
// the catalog (§348), handed to Reserve by the caller.
const (
// ReservedState is the builder's old state section of the reservations.
ReservedState = "reserved"
// ReservedBucket is its bucket.
ReservedBucket = "mesh-delivery-builder_" + ReservedState
// ReservedTool is builder_reserved, ReservedDropTool builder_reserved_drop.
ReservedTool = "builder_reserved"
ReservedDropTool = "builder_reserved_drop"
// UnreleasedBound is the publish's bound: past it a built reservation with an artifact unpublished is the finding
// version-unreleased (§369).
UnreleasedBound = 10 * time.Minute
// VersionUnreleased is that finding.
VersionUnreleased = "version-unreleased"
)
// ReservedVersion is the value of a reservation key.
type ReservedVersion struct {
V int `json:"v"`
Module string `json:"module"`
Series string `json:"series"` // MAJOR.MINOR
Version string `json:"version"` // MAJOR.MINOR.PATCH
// Artifacts are the module's artifacts at the version: the digest once built, empty before.
Artifacts map[string]string `json:"artifacts"`
// BuiltAt is when the build ended with every digest known; the publish's bound counts from it.
BuiltAt time.Time `json:"built-at,omitzero"`
// Published are the artifacts a registry seat said published at this version with the built digest.
Published map[string]bool `json:"published,omitempty"`
At time.Time `json:"at"`
}
// ReservedKey is the key of module's reservation of series (MAJOR.MINOR).
func ReservedKey(module, series string) string { return Key(module, series) }
// ErrNotFilled is what Reserve answers before the catalog says versions-filled: nothing is reserved until then (§369).
var ErrNotFilled = errors.New("versions-filled is not heard: nothing is reserved before the catalog's versions are filled")
// ReserveStore is the reservations' bucket: compare-and-set, and deletion.
type ReserveStore interface {
Store
Delete(ctx context.Context, key string) error
}
// Reserver reserves versions.
type Reserver struct {
Store ReserveStore
// ReleasedOf are the patch versions (MAJOR.MINOR.PATCH) the catalog holds released for module, from its verb versions.
ReleasedOf func(ctx context.Context, module string) ([]string, error)
// Filled says the catalog said versions-filled.
Filled func() bool
Now func() time.Time
}
func (r *Reserver) now() time.Time {
if r.Now != nil {
return r.Now().UTC()
}
return time.Now().UTC()
}
// SplitVersion is a MAJOR.MINOR.PATCH version's series and patch.
func SplitVersion(version string) (series string, patch int, err error) {
i := strings.LastIndex(version, ".")
if i < 0 || strings.Count(version, ".") != 2 {
return "", 0, fmt.Errorf("%q is not MAJOR.MINOR.PATCH", version)
}
patch, err = strconv.Atoi(version[i+1:])
if err != nil || patch < 0 {
return "", 0, fmt.Errorf("%q is not MAJOR.MINOR.PATCH", version)
}
return version[:i], patch, nil
}
// Reserve reserves the next version of module's series for a build that starts, for artifacts (their names), and answers
// it. The next patch is the one after the higher of the highest released and the one reserved, 0 when there is neither. A
// conflicting write (another build reserved first) is read again and the next taken.
func (r *Reserver) Reserve(ctx context.Context, module, series string, artifacts []string) (ReservedVersion, error) {
if r.Filled != nil && !r.Filled() {
return ReservedVersion{}, ErrNotFilled
}
released, err := r.ReleasedOf(ctx, module)
if err != nil {
return ReservedVersion{}, fmt.Errorf("the released versions of %s cannot be read: %w", module, err)
}
high := -1
for _, v := range released {
if s, p, err := SplitVersion(v); err == nil && s == series && p > high {
high = p
}
}
key := ReservedKey(module, series)
for {
cur, rev, found, err := r.get(ctx, key)
if err != nil {
return ReservedVersion{}, err
}
next := high
if found {
if _, p, err := SplitVersion(cur.Version); err == nil && p > next {
next = p
}
}
out := ReservedVersion{V: 1, Module: module, Series: series, Version: fmt.Sprintf("%s.%d", series, next+1), Artifacts: map[string]string{}, At: r.now()}
for _, a := range artifacts {
out.Artifacts[a] = ""
}
raw, _ := json.Marshal(out)
if found {
_, err = r.Store.Update(ctx, key, raw, rev)
} else {
_, err = r.Store.Create(ctx, key, raw)
}
if errors.Is(err, ErrConflict) {
continue
}
if err != nil {
return ReservedVersion{}, fmt.Errorf("%s cannot be reserved: %w", key, err)
}
return out, nil
}
}
func (r *Reserver) get(ctx context.Context, key string) (ReservedVersion, uint64, bool, error) {
raw, rev, found, err := r.Store.Get(ctx, key)
if err != nil || !found {
return ReservedVersion{}, 0, false, err
}
var v ReservedVersion
if err := json.Unmarshal(raw, &v); err != nil {
return ReservedVersion{}, 0, false, fmt.Errorf("%s holds what is not a reservation: %w", key, err)
}
return v, rev, true, nil
}
// Built records artifact's digest under the reserved version of module's series; once every artifact has one the build is
// ended (BuiltAt). A reservation moved on to another version (a newer build took the next) is left alone and answers false.
func (r *Reserver) Built(ctx context.Context, module, version, artifact, digest string) (bool, error) {
series, _, err := SplitVersion(version)
if err != nil {
return false, err
}
key := ReservedKey(module, series)
for {
cur, rev, found, err := r.get(ctx, key)
if err != nil || !found || cur.Version != version {
return false, err
}
cur.Artifacts[artifact] = digest
if cur.BuiltAt.IsZero() && allSet(cur.Artifacts) {
cur.BuiltAt = r.now()
}
raw, _ := json.Marshal(cur)
if _, err := r.Store.Update(ctx, key, raw, rev); errors.Is(err, ErrConflict) {
continue
} else if err != nil {
return false, fmt.Errorf("%s cannot be written: %w", key, err)
}
return true, nil
}
}
func allSet(m map[string]string) bool {
for _, d := range m {
if d == "" {
return false
}
}
return true
}
// Released takes the catalog's version-released in: the key of that version is deleted, and every key of module's other
// series (the manifest moved on). A key holding a newer reserved version of the same series stays. It answers the keys
// deleted.
func (r *Reserver) Released(ctx context.Context, module, version string) ([]string, error) {
series, _, err := SplitVersion(version)
if err != nil {
return nil, err
}
keys, err := r.Store.KeysOf(ctx, NameFilter(module))
if err != nil {
return nil, fmt.Errorf("the reservations of %s cannot be listed: %w", module, err)
}
var gone []string
for _, k := range keys {
cur, _, found, err := r.get(ctx, k)
if err != nil {
return gone, err
}
if !found || (cur.Series == series && cur.Version != version) {
continue
}
if err := r.Store.Delete(ctx, k); err != nil {
return gone, fmt.Errorf("%s cannot be deleted: %w", k, err)
}
gone = append(gone, k)
}
sort.Strings(gone)
return gone, nil
}
// Unreleased are the reservations whose build ended more than UnreleasedBound before now with an artifact unpublished:
// the finding version-unreleased, naming the module, the version and the artifacts missing (published says whether an
// artifact is). A reservation whose build has not ended is not one.
type Unreleased struct {
Module, Version string
Missing []string
}
// Overdue answers the reservations past the bound with an artifact not published.
func (r *Reserver) Overdue(ctx context.Context, published func(module, version, artifact string) bool) ([]Unreleased, error) {
keys, err := r.Store.KeysOf(ctx, AllFilter)
if err != nil {
return nil, fmt.Errorf("the reservations cannot be listed: %w", err)
}
sort.Strings(keys)
var out []Unreleased
for _, k := range keys {
cur, _, found, err := r.get(ctx, k)
if err != nil {
return nil, err
}
if !found || cur.BuiltAt.IsZero() || r.now().Sub(cur.BuiltAt) < UnreleasedBound {
continue
}
var missing []string
for _, a := range cur.Missing() {
if published == nil || !published(cur.Module, cur.Version, a) {
missing = append(missing, a)
}
}
if len(missing) > 0 {
out = append(out, Unreleased{Module: cur.Module, Version: cur.Version, Missing: missing})
}
}
return out, nil
}
// All are the reservations, sorted by key: builder_reserved.
func (r *Reserver) All(ctx context.Context) ([]ReservedVersion, error) {
keys, err := r.Store.KeysOf(ctx, AllFilter)
if err != nil {
return nil, fmt.Errorf("the reservations cannot be listed: %w", err)
}
sort.Strings(keys)
var out []ReservedVersion
for _, k := range keys {
cur, _, found, err := r.get(ctx, k)
if err != nil {
return nil, err
}
if found {
out = append(out, cur)
}
}
return out, nil
}
// Drop deletes module's reservation of series on a person's word (builder_reserved_drop); the caller writes the hand-act
// log. It answers whether there was one.
func (r *Reserver) Drop(ctx context.Context, module, series string) (bool, error) {
key := ReservedKey(module, series)
_, _, found, err := r.get(ctx, key)
if err != nil || !found {
return false, err
}
return true, r.Store.Delete(ctx, key)
}
// Published takes a registry seat's published in: the artifacts of a reservation at version (a leading v is the Go form)
// whose built digest it names are marked published. It answers the keys it marked.
func (r *Reserver) Published(ctx context.Context, version, digest string) ([]string, error) {
version = strings.TrimPrefix(version, "v")
if version == "" || digest == "" {
return nil, nil
}
keys, err := r.Store.KeysOf(ctx, AllFilter)
if err != nil {
return nil, fmt.Errorf("the reservations cannot be listed: %w", err)
}
sort.Strings(keys)
var marked []string
for _, k := range keys {
again:
cur, rev, found, err := r.get(ctx, k)
if err != nil {
return marked, err
}
if !found || cur.Version != version {
continue
}
changed := false
for a, d := range cur.Artifacts {
if d == digest && !cur.Published[a] {
if cur.Published == nil {
cur.Published = map[string]bool{}
}
cur.Published[a], changed = true, true
}
}
if !changed {
continue
}
raw, _ := json.Marshal(cur)
if _, err := r.Store.Update(ctx, k, raw, rev); errors.Is(err, ErrConflict) {
goto again
} else if err != nil {
return marked, fmt.Errorf("%s cannot be written: %w", k, err)
}
marked = append(marked, k)
}
return marked, nil
}
// Of is module's reservation of series, and whether there is one; with an empty series the module's highest.
func (r *Reserver) Of(ctx context.Context, module, version string) (ReservedVersion, bool, error) {
all, err := r.All(ctx)
if err != nil {
return ReservedVersion{}, false, err
}
var out ReservedVersion
found := false
for _, v := range all {
if v.Module != module || (version != "" && v.Version != version) {
continue
}
if !found {
out, found = v, true
continue
}
_, p, _ := SplitVersion(v.Version)
_, q, _ := SplitVersion(out.Version)
if v.Series > out.Series || (v.Series == out.Series && p > q) {
out = v
}
}
return out, found, nil
}
const reservedBoundPrefix = "reserved."
// ReservedBoundKey is the bound of a reservation's publish in the state bounds (key-ttl): `reserved.<reserved key>`, armed
// when its build ended with every digest known. Its end while the reservation stands is the finding version-unreleased.
func ReservedBoundKey(module, series string) string {
return ReservedBoundFor(ReservedKey(module, series))
}
// ReservedBoundFor is the bound of the reservation under key.
func ReservedBoundFor(key string) string { return reservedBoundPrefix + key }
// ReservedBoundOf is the reserved key a reservation's bound names, and whether key is one.
func ReservedBoundOf(key string) (string, bool) { return strings.CutPrefix(key, reservedBoundPrefix) }
// readOnlyReserve is the reservations read by the tools bundle, which writes nothing.
type readOnlyReserve struct{ readStore }
func (readOnlyReserve) Delete(context.Context, string) error { return errReadOnly }
// AnswerReserved is builder_reserved: every reservation, or one module's.
func AnswerReserved(r Reader, module string) (map[string]any, error) {
rs := &Reserver{Store: readOnlyReserve{readStore{r}}}
all, err := rs.All(context.Background())
if err != nil {
return nil, err
}
out := []ReservedVersion{}
for _, v := range all {
if module == "" || v.Module == module {
out = append(out, v)
}
}
return map[string]any{"reserved": out}, nil
}
// AnswerReservedVerb is the seat mesh-builds' verb reserved: module's reserved version (the given one, else the highest)
// with each artifact's digest, an artifact not built yet with none.
func AnswerReservedVerb(r Reader, module, version string) (meshbuilds.ReservedAnswer, error) {
module = strings.TrimSpace(module)
if module == "" {
return meshbuilds.ReservedAnswer{}, errors.New("module is required")
}
rs := &Reserver{Store: readOnlyReserve{readStore{r}}}
v, found, err := rs.Of(context.Background(), module, strings.TrimSpace(version))
if err != nil {
return meshbuilds.ReservedAnswer{}, err
}
if !found {
return meshbuilds.ReservedAnswer{}, fmt.Errorf("%s has no reserved version%s", module, map[bool]string{true: " " + version, false: ""}[version != ""])
}
out := meshbuilds.ReservedAnswer{Module: v.Module, Version: v.Version, Artifacts: []meshbuilds.ReservedAnswerArtifactsItem{}}
names := make([]string, 0, len(v.Artifacts))
for a := range v.Artifacts {
names = append(names, a)
}
sort.Strings(names)
for _, a := range names {
out.Artifacts = append(out.Artifacts, meshbuilds.ReservedAnswerArtifactsItem{Artifact: a, Digest: v.Artifacts[a]})
}
return out, nil
}
// The descriptions of the reservation's tools.
const (
ReservedDescription = "The versions the builder reserved (its state reserved, design 47 G section 369): one per module and MAJOR.MINOR, " +
"the version its build reserved at its start, each artifact's digest once built, when the build ended and which artifacts a registry " +
"seat said published. With module, that module's only. A key goes when the catalog says version-released for it. Read-only."
ReservedDropDescription = "Drop a module's reservation of one MAJOR.MINOR by hand (design 47 G section 369), when its version will never be " +
"released; the next build takes the next after the highest released. A hand act: it is recorded in the hand-act log."
// ReservedVerb is the seat mesh-builds' verb served as a tool of the same name.
ReservedVerb = meshbuilds.VerbReserved
)
// Missing are the artifacts of v no registry seat said published, sorted.
func (v ReservedVersion) Missing() []string {
var out []string
for a := range v.Artifacts {
if !v.Published[a] {
out = append(out, a)
}
}
sort.Strings(out)
return out
}
// At is the reservation under key, and whether there is one.
func (r *Reserver) At(ctx context.Context, key string) (ReservedVersion, bool, error) {
v, _, found, err := r.get(ctx, key)
return v, found, err
}
@@ -0,0 +1,158 @@
package versions
import (
"encoding/json"
"context"
"errors"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
"reflect"
"testing"
"time"
)
type reserveMem struct {
*mem
onUpdate func() // runs once before the next Update
}
func (m *reserveMem) Delete(_ context.Context, key string) error {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.vals, key)
delete(m.revs, key)
return nil
}
func (m *reserveMem) Update(ctx context.Context, key string, value []byte, rev uint64) (uint64, error) {
if f := m.onUpdate; f != nil {
m.onUpdate = nil
f()
}
return m.mem.Update(ctx, key, value, rev)
}
// TestReservation: design 47 G §369 (test 102): two builds close together reserve two versions, the key goes on
// version-released, a built reservation with an artifact unpublished past the bound is unreleased, and nothing is reserved
// before versions-filled.
func TestReservation(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
released := []string{"0.1.18", "0.1.19", "0.2.7"}
filled := true
store := &reserveMem{mem: newMem()}
r := &Reserver{Store: store, Filled: func() bool { return filled }, Now: func() time.Time { return now },
ReleasedOf: func(context.Context, string) ([]string, error) { return released, nil }}
arts := []string{"m:a", "m:b"}
filled = false
if _, err := r.Reserve(ctx, "m", "0.1", arts); !errors.Is(err, ErrNotFilled) {
t.Fatalf("before versions-filled: %v, want ErrNotFilled", err)
}
if got, _ := r.All(ctx); len(got) != 0 {
t.Fatalf("a reservation was written before the fill: %v", got)
}
filled = true
first, err := r.Reserve(ctx, "m", "0.1", arts)
if err != nil || first.Version != "0.1.20" {
t.Fatalf("first build reserves %q (%v), want 0.1.20 after the highest released 0.1.19", first.Version, err)
}
second, err := r.Reserve(ctx, "m", "0.1", arts)
if err != nil || second.Version != "0.1.21" {
t.Fatalf("second build reserves %q (%v), want 0.1.21: never the first's", second.Version, err)
}
// Two builds at once: the rival reserves between this one's read and write, and this one takes the next.
store.onUpdate = func() {
if _, err := r.Reserve(ctx, "m", "0.1", arts); err != nil {
t.Error(err)
}
}
third, err := r.Reserve(ctx, "m", "0.1", arts)
if err != nil || third.Version != "0.1.23" {
t.Fatalf("a build that lost the compare-and-set reserves %q (%v), want 0.1.23", third.Version, err)
}
if v, _ := r.Reserve(ctx, "m", "0.3", arts); v.Version != "0.3.0" {
t.Errorf("a series with nothing released reserves %q, want 0.3.0", v.Version)
}
// The artifacts' digests, then the bound.
for _, a := range arts {
if ok, err := r.Built(ctx, "m", "0.1.23", a, "sha256:"+a); err != nil || !ok {
t.Fatalf("built %s: %v %v", a, ok, err)
}
}
if ok, _ := r.Built(ctx, "m", "0.1.22", "m:a", "x"); ok {
t.Error("a digest was written under a version a newer build superseded")
}
published := func(_, _, a string) bool { return a == "m:a" }
if got, _ := r.Overdue(ctx, published); len(got) != 0 {
t.Fatalf("unreleased before the bound: %v", got)
}
now = now.Add(UnreleasedBound)
got, err := r.Overdue(ctx, published)
want := []Unreleased{{Module: "m", Version: "0.1.23", Missing: []string{"m:b"}}}
if err != nil || !reflect.DeepEqual(got, want) {
t.Fatalf("past the bound: %v (%v), want %v", got, err, want)
}
// version-released: its key goes, and the other series' keys of the module (the manifest moved on).
if gone, err := r.Released(ctx, "m", "0.1.23"); err != nil || len(gone) != 2 {
t.Fatalf("released 0.1.23 deleted %v (%v), want the 0.1 key and the 0.3 key", gone, err)
}
if all, _ := r.All(ctx); len(all) != 0 {
t.Errorf("reservations left: %v", all)
}
if again, _ := r.Reserve(ctx, "m", "0.1", arts); again.Version != "0.1.20" {
t.Logf("with the catalog list unchanged the next is %s", again.Version)
}
if ok, _ := r.Drop(ctx, "m", "0.1"); !ok {
t.Error("builder_reserved_drop found no reservation")
}
}
// TestReservationIsReadAndMarkedPublished: design 47 G §369 — the reservation is read by builder_reserved and by the seat's
// verb reserved with each artifact's digest, a registry seat's published marks the artifacts of that version and digest, and
// the missing artifacts it names are the unpublished ones.
func TestReservationIsReadAndMarkedPublished(t *testing.T) {
ctx := context.Background()
store := &reserveMem{mem: newMem()}
r := &Reserver{Store: store, ReleasedOf: func(context.Context, string) ([]string, error) { return []string{"1.4.2"}, nil }}
v, err := r.Reserve(ctx, "m", "1.4", []string{"m:a", "m:b"})
if err != nil || v.Version != "1.4.3" {
t.Fatalf("reserved %q (%v), want 1.4.3", v.Version, err)
}
if _, err := r.Built(ctx, "m", "1.4.3", "m:a", "sha256:aa"); err != nil {
t.Fatal(err)
}
rd := Reader{
Keys: func() ([]string, error) { return store.KeysOf(ctx, AllFilter) },
Get: func(key string) (json.RawMessage, bool, error) {
raw, _, ok, err := store.Get(ctx, key)
return raw, ok, err
}}
ans, err := AnswerReservedVerb(rd, "m", "")
want := meshbuilds.ReservedAnswer{Module: "m", Version: "1.4.3", Artifacts: []meshbuilds.ReservedAnswerArtifactsItem{{Artifact: "m:a", Digest: "sha256:aa"}, {Artifact: "m:b"}}}
if err != nil || !reflect.DeepEqual(ans, want) {
t.Fatalf("reserved answers %+v (%v), want %+v", ans, err, want)
}
if _, err := AnswerReservedVerb(rd, "m", "1.4.9"); err == nil {
t.Error("a version never reserved was answered")
}
if got, err := AnswerReserved(rd, "other"); err != nil || len(got["reserved"].([]ReservedVersion)) != 0 {
t.Errorf("builder_reserved of another module = %v (%v), want none", got, err)
}
if marked, err := r.Published(ctx, "v1.4.3", "sha256:aa"); err != nil || len(marked) != 1 {
t.Fatalf("a published of 1.4.3 with a built digest marked %v (%v)", marked, err)
}
if marked, _ := r.Published(ctx, "1.4.3", "sha256:other"); len(marked) != 0 {
t.Errorf("a published with another digest marked %v", marked)
}
cur, _, _ := r.At(ctx, ReservedKey("m", "1.4"))
if got := cur.Missing(); !reflect.DeepEqual(got, []string{"m:b"}) {
t.Errorf("missing = %v, want m:b", got)
}
if key, ok := ReservedBoundOf(ReservedBoundKey("m", "1.4")); !ok || key != ReservedKey("m", "1.4") || ReservedBoundFor(key) != ReservedBoundKey("m", "1.4") {
t.Errorf("the bound key does not name the reservation back: %q %v", key, ok)
}
}
+17 -8
View File
@@ -38,9 +38,14 @@ type artifact struct {
}
type manifest struct {
Module string `json:"module"`
Version string `json:"version"`
Slug string `json:"slug"`
Module string `json:"module"`
Version string `json:"version"`
Slug string `json:"slug"`
Claims []struct {
Name string `json:"name"`
Scope string `json:"scope"`
Serves []string `json:"serves"`
} `json:"claims"`
Tools []string `json:"tools"`
Emits []string `json:"emits"`
Consumes []string `json:"consumes"`
@@ -97,7 +102,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if m.Module != "mesh-delivery-builder" {
t.Errorf("module = %q", m.Module)
}
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory
t.Errorf("state = %+v, want %+v", m.State, want)
}
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
@@ -127,7 +132,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// The watchdogs seat's presence says a node's machine is heard (issue 1966, ADR 0596 §2, ADR 0668 §3).
if want := []string{"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-catalog.versions", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
}
@@ -139,7 +144,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// DLV-M1: the indexer's and the assigner's views families of the states it follows.
if want := []string{"container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed",
"go-package-registry.kept", "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed",
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted",
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled",
"mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.assignments", "mesh-delivery-checker.tag-wanted", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.packages", "node-engine.manifest-reads-changed",
@@ -155,7 +160,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
}
// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
if want := []string{"builder_announce_again", "builder_ask_again", "builder_bounds", "builder_build_again", "builder_builds_since", "builder_engine_reads", "builder_first_view_again", "builder_first_views",
"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions"}; !reflect.DeepEqual(sorted(m.Tools), want) {
"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// Design 51 *Events of the builder*: numbered and old-form-closed on the family version, apart from the builds' ends.
@@ -166,7 +171,11 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if len(m.Logs) != 1 || m.Logs[0].Name != builds.BuilderLog || m.Logs[0].MaxAgeDays != builds.BuilderLogDays || m.Logs[0].MaxMiB != builds.BuilderLogMiB {
t.Errorf("logs = %+v, want %s kept %d days in %d MiB", m.Logs, builds.BuilderLog, builds.BuilderLogDays, builds.BuilderLogMiB)
}
for _, k := range []string{"streams", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
// Design 47 G §52, §369: the seat mesh-builds, held once for the mesh, serving reserved.
if len(m.Claims) != 1 || m.Claims[0].Name != "mesh-builds" || m.Claims[0].Scope != "mesh" || !reflect.DeepEqual(m.Claims[0].Serves, []string{"reserved"}) {
t.Errorf("claims = %+v, want mesh-builds, scope mesh, serving reserved", m.Claims)
}
for _, k := range []string{"streams", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the builder has none", k)
}
+16
View File
@@ -2,6 +2,15 @@
"module": "mesh-delivery-builder",
"version": "1",
"slug": "builder",
"claims": [
{
"name": "mesh-builds",
"scope": "mesh",
"serves": [
"reserved"
]
}
],
"tools": [
"builder_announce_again",
"builder_ask_again",
@@ -13,12 +22,15 @@
"builder_floor_waits",
"builder_old_form",
"builder_publish_again",
"builder_reserved",
"builder_reserved_drop",
"builder_seat_queue",
"builder_stuck",
"builder_tag",
"builder_tag_asks",
"builder_untag",
"builder_versions",
"reserved",
"builder_first_views",
"builder_first_view_again"
],
@@ -40,6 +52,8 @@
"mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported",
"mesh-catalog.manifest-accepted",
"mesh-catalog.version-released",
"mesh-catalog.versions-filled",
"mesh-conditions.condition-cleared",
"mesh-delivery-checker.tag-wanted",
"mesh-delivery-forge-bridge.forge.**",
@@ -66,6 +80,7 @@
"seat:go-package-registry.remove",
"seat:go-package-registry.versions",
"seat:mesh-artifact-sweep.held",
"seat:mesh-catalog.versions",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-watchdogs.presence",
"seat:node-build-go.drop",
@@ -92,6 +107,7 @@
{ "name": "old-form", "history": 1 },
{ "name": "bounds", "history": 1, "key-ttl": true },
{ "name": "queues", "history": 1 },
{ "name": "reserved", "history": 1 },
{ "name": "reasks", "history": 1 },
{ "name": "tag-ask-bounds", "history": 1, "key-ttl": true },
{ "name": "tag-asks", "history": 1 },
@@ -5,6 +5,7 @@ import (
"errors"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-sdk/go/keys"
@@ -191,4 +192,37 @@ func TestForge(t *testing.T) {
}
}
})
// Design 47 G §164: an owed act's outcome names its seat, its id, one outcome and words for every outcome but
// performed; an act the bridge cannot decode may name no id.
t.Run("owed ended", func(t *testing.T) {
at := time.Date(2026, 11, 9, 20, 0, 0, 0, time.UTC)
ok := OwedEnded{Of: "mesh-delivery", ID: "g1.view.1", Outcome: OwedPerformed, At: at}
raw, err := EncodeOwedEnded(ok)
if back, derr := DecodeOwedEnded(raw); err != nil || derr != nil || back.V != Version || back.ID != ok.ID {
t.Fatalf("round trip: %s %v %v", raw, err, derr)
}
if _, err := DecodeOwedEnded([]byte(`{"v":99,"of":"x","id":"a","outcome":"performed","at":"2026-11-09T20:00:00Z"}`)); err == nil {
t.Errorf("an unknown version was read")
}
for _, c := range []struct {
name string
edit func(*OwedEnded)
want string
}{
{"no seat", func(e *OwedEnded) { e.Of = "" }, "no seat"},
{"an outcome of no kind", func(e *OwedEnded) { e.Outcome = "done" }, "none of"},
{"no id", func(e *OwedEnded) { e.ID = "" }, "no id"},
{"a refusal with no words", func(e *OwedEnded) { e.Outcome = OwedRefused }, "no words"},
{"dropped with no reason", func(e *OwedEnded) { e.Outcome = OwedDropped }, "no words"},
{"no time", func(e *OwedEnded) { e.At = time.Time{} }, "no time"},
{"undecodable with no id is said", func(e *OwedEnded) { e.ID, e.Outcome, e.Words = "", OwedUndecodable, "not json" }, ""},
} {
e := ok
c.edit(&e)
_, err := EncodeOwedEnded(e)
if (c.want == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.want)) {
t.Errorf("%s: %v, want %q", c.name, err, c.want)
}
}
})
}
@@ -0,0 +1,76 @@
package forgemsg
import (
"encoding/json"
"fmt"
"slices"
"time"
)
// OwedEndedEvent is the name of an owed act's outcome on the forge bridge's stream `forge` (design 47 G §164): the
// bridge says it for each act the gate or the checker said as the event `owed`. Its two parts are no repository's
// forge event (`forge.<owner>.<name>.<kind>`).
const OwedEndedEvent = "owed.ended"
// The outcomes of an owed act (design 47 G §164, §168): performed, refused (the forge's own words), could not be
// performed (the forge did not answer within the call's bound, with the error), undecodable (an act the bridge cannot
// read, with why) and, from the tool `bridge_owed_drop`, dropped.
const (
OwedPerformed = "performed"
OwedRefused = "refused"
OwedCouldNotBePerformed = "could not be performed"
OwedUndecodable = "undecodable"
OwedDropped = "dropped"
)
// OwedOutcomes are the outcomes an owed.ended says, one of them.
var OwedOutcomes = []string{OwedPerformed, OwedRefused, OwedCouldNotBePerformed, OwedUndecodable, OwedDropped}
// OwedEnded is an owed act's outcome (design 47 G §164): the act's id and one outcome, with the words of any outcome
// but performed. Of is the seat that said the act (mesh-delivery for the gate's, mesh-delivery-checker for the
// checker's), so each hears its own ids (§164). An act the bridge cannot decode may name no id.
type OwedEnded struct {
V int `json:"v"`
Of string `json:"of"`
ID string `json:"id,omitempty"`
Outcome string `json:"outcome"`
Words string `json:"words,omitempty"`
At time.Time `json:"at"`
}
func (e OwedEnded) check() error {
switch {
case e.Of == "":
return fmt.Errorf("an owed act's outcome names no seat that said it")
case !slices.Contains(OwedOutcomes, e.Outcome):
return fmt.Errorf("the outcome %q of an owed act is none of %v", e.Outcome, OwedOutcomes)
case e.ID == "" && e.Outcome != OwedUndecodable:
return fmt.Errorf("the outcome %s of an owed act names no id", e.Outcome)
case e.Outcome != OwedPerformed && e.Words == "":
return fmt.Errorf("the outcome %s of the owed act %s says no words", e.Outcome, e.ID)
case e.At.IsZero():
return fmt.Errorf("the outcome of the owed act %s says no time", e.ID)
}
return nil
}
// EncodeOwedEnded is e as JSON at Version; it refuses what check refuses.
func EncodeOwedEnded(e OwedEnded) ([]byte, error) {
if err := e.check(); err != nil {
return nil, err
}
e.V = Version
return json.Marshal(e)
}
// DecodeOwedEnded reads an owed act's outcome; it refuses an unknown version and what EncodeOwedEnded refuses.
func DecodeOwedEnded(data []byte) (OwedEnded, error) {
var e OwedEnded
if err := decode(data, &e); err != nil {
return OwedEnded{}, err
}
if err := e.check(); err != nil {
return OwedEnded{}, err
}
return e, nil
}