Merge branch 'feat/47-G-build' into feat/47-G-build-bridge
This commit is contained in:
@@ -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
|
||||
)
|
||||
|
||||
@@ -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, ®)
|
||||
}
|
||||
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, ®); 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, ®) == 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 },
|
||||
|
||||
Reference in New Issue
Block a user