Merge branch 'feat/47-G-build' into feat/47-G-build-bridge
This commit is contained in:
@@ -0,0 +1,176 @@
|
||||
package meshwantsclient
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
// RevisionVerb is the seat verb a part takes its first view of the assignment half by, as its manifest's invokes names
|
||||
// it.
|
||||
const RevisionVerb = "seat:" + seat.Seat + "." + seat.VerbRevision
|
||||
|
||||
// kindTrunk is the kind of an assignment graph revision a running part acts on; a candidate's moves nothing.
|
||||
const kindTrunk = "trunk"
|
||||
|
||||
// Call asks one seat verb over the bus and answers what the holder said, whatever wraps it (process.Conn.Call, or
|
||||
// stdio.Ask through the tool runner).
|
||||
type Call func(key string, body any) (json.RawMessage, error)
|
||||
|
||||
// assignmentCreated stands for the view's Created in its follower: a revision view is never made again, so one fixed
|
||||
// time, never zero, lets the follower tell a revision heard from one it holds.
|
||||
var assignmentCreated = time.Unix(0, 0).UTC()
|
||||
|
||||
// assignmentViewName is how an assignment half's revision is named in a part's first-view status (views.Status.View,
|
||||
// WaitsFor).
|
||||
func assignmentViewName(revision int64) string { return strconv.FormatInt(revision, 10) }
|
||||
|
||||
// Assignment is a part's view of the assignment half (design 47 G §245): a first view of the newest trunk revision asked
|
||||
// of the seat mesh-wants' verb `revision`, then kept from its event `assignment-revision`, built on go/views' Follower,
|
||||
// so the runtime's tools `<module>_first_views` and `<module>_first_view_again` read and steer it as every first view.
|
||||
// Its picture is keyed `<node>.<module>`, each value that entry's reference. With no view the part acts on nothing:
|
||||
// Revision says false, and a revision heard and not held yet is waited for, said in the status, and past
|
||||
// views.WaitsBound raised as view-waits.
|
||||
type Assignment struct {
|
||||
Follower *views.Follower
|
||||
// Held, when set, is told of each revision once the view holds it, on the ask's goroutine and outside any lock.
|
||||
Held func(seat.AssignmentRevision)
|
||||
|
||||
call Call
|
||||
mu sync.Mutex
|
||||
held *seat.AssignmentRevision
|
||||
waits int64
|
||||
}
|
||||
|
||||
// NewAssignment is an Assignment asking the seat through call. Nothing is asked before Start.
|
||||
func NewAssignment(call Call) *Assignment {
|
||||
a := &Assignment{call: call}
|
||||
a.Follower = &views.Follower{Owner: seat.Seat, Tool: RevisionVerb, State: seat.StateAssignmentRevisions, Ask: a.ask}
|
||||
return a
|
||||
}
|
||||
|
||||
// WireAssignment is an Assignment adopted by a stdio part's views.Reader: the reader writes its conditions and its
|
||||
// view-waits finding, reads it as `mesh-wants.assignment-revisions`, lists and asks it again in its first-view tools, and
|
||||
// at its Start subscribes the seat's event assignment-revision (the part's manifest consumes it) into Hear and asks the
|
||||
// first view. held, when not nil, is the Assignment's Held. The part declares the seat's verb revision in invokes.
|
||||
func WireAssignment(r *views.Reader, call Call, held func(seat.AssignmentRevision)) (*Assignment, error) {
|
||||
a := NewAssignment(call)
|
||||
a.Held = held
|
||||
if err := r.Adopt(a.Follower, seat.Seat, seat.EventAssignmentRevision, a.Hear); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// WireAssignmentProcess is WireAssignment for a process's views.ProcessReader: its picture
|
||||
// (`mesh-wants_assignment-revisions`) reads the half as a read-only bucket, and the seat's event assignment-revision on
|
||||
// the module's durable consumer goes to Hear.
|
||||
func WireAssignmentProcess(r *views.ProcessReader, call Call, held func(seat.AssignmentRevision)) (*Assignment, *views.Picture, error) {
|
||||
a := NewAssignment(call)
|
||||
a.Held = held
|
||||
p, err := r.Adopt(a.Follower, seat.Seat, seat.EventAssignmentRevision, a.Hear)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return a, p, nil
|
||||
}
|
||||
|
||||
// Start asks the first view, the newest trunk revision, until it is answered or ctx ends.
|
||||
func (a *Assignment) Start(ctx context.Context) error { return a.Follower.Start(ctx) }
|
||||
|
||||
// Revision is the assignment graph revision the view holds; false while it has none, when the part acts on nothing new.
|
||||
func (a *Assignment) Revision() (seat.AssignmentRevision, bool) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if a.held == nil {
|
||||
return seat.AssignmentRevision{}, false
|
||||
}
|
||||
return *a.held, true
|
||||
}
|
||||
|
||||
// Hear takes one `assignment-revision` event, as the part's durable consumer of the seat's events hands it. A
|
||||
// candidate's revision moves nothing; a trunk revision the view holds, or one older, is passed over; a newer one is
|
||||
// waited for and asked on its own goroutine, so the consumer is never held while the holder is away. Data that is no
|
||||
// such event is an error.
|
||||
func (a *Assignment) Hear(ctx context.Context, data []byte) error {
|
||||
var ev seat.AssignmentRevisionEvent
|
||||
if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 {
|
||||
return fmt.Errorf("meshwantsclient: not a %s event: %s", seat.EventAssignmentRevision, clipBytes(data))
|
||||
}
|
||||
if ev.Kind != kindTrunk {
|
||||
return nil
|
||||
}
|
||||
a.mu.Lock()
|
||||
held := int64(0)
|
||||
if a.held != nil {
|
||||
held = a.held.Revision
|
||||
}
|
||||
if ev.Revision > a.waits {
|
||||
a.waits = ev.Revision
|
||||
}
|
||||
a.mu.Unlock()
|
||||
if ev.Revision <= held {
|
||||
return nil
|
||||
}
|
||||
// The follower keeps the newest revision heard, so an ask that was running when it was heard is asked once more.
|
||||
a.Follower.Apply(views.Event{Owner: seat.Seat, State: seat.StateAssignmentRevisions, Revision: uint64(ev.Revision), Created: assignmentCreated, Gap: true})
|
||||
a.Follower.WaitFor(ctx, assignmentViewName(ev.Revision))
|
||||
go func() {
|
||||
_ = a.Follower.Gap(ctx, fmt.Sprintf("assignment revision %d said, the view holds %d", ev.Revision, held))
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// ask is the Follower's Ask: the newest trunk revision, whole in one answer (its entries are named by digest).
|
||||
func (a *Assignment) ask(_ context.Context, _ string) (views.Answer, error) {
|
||||
raw, err := a.call(RevisionVerb, seat.RevisionInput{})
|
||||
if err != nil {
|
||||
return views.Answer{}, err
|
||||
}
|
||||
answer, said, ok := buscalls.Unwrap(raw)
|
||||
if !ok {
|
||||
return views.Answer{}, fmt.Errorf("%s answered: %s", RevisionVerb, said)
|
||||
}
|
||||
var ans seat.RevisionAnswer
|
||||
if err := json.Unmarshal(answer, &ans); err != nil {
|
||||
return views.Answer{}, fmt.Errorf("%s answered what is not its answer: %s", RevisionVerb, clipBytes(answer))
|
||||
}
|
||||
r := ans.Revision
|
||||
if r.Revision <= 0 || r.Kind != kindTrunk {
|
||||
return views.Answer{}, fmt.Errorf("%s answered a %q revision %d for the trunk", RevisionVerb, r.Kind, r.Revision)
|
||||
}
|
||||
values := make(map[string]json.RawMessage, len(r.Entries))
|
||||
for _, e := range r.Entries {
|
||||
v, _ := json.Marshal(e)
|
||||
values[e.Node+"."+e.Module] = v
|
||||
}
|
||||
a.mu.Lock()
|
||||
moved := a.held == nil || a.held.Revision != r.Revision
|
||||
a.held = &r
|
||||
holds := a.waits <= r.Revision
|
||||
if holds {
|
||||
a.waits = 0
|
||||
}
|
||||
a.mu.Unlock()
|
||||
if holds {
|
||||
a.Follower.Holds(assignmentViewName(r.Revision))
|
||||
}
|
||||
if moved && a.Held != nil {
|
||||
a.Held(r)
|
||||
}
|
||||
return views.Answer{V: 1, Owner: seat.Seat, State: seat.StateAssignmentRevisions, Revision: uint64(r.Revision), Created: assignmentCreated, Values: values, At: time.Now()}, nil
|
||||
}
|
||||
|
||||
func clipBytes(b []byte) string {
|
||||
if len(b) > 200 {
|
||||
return string(b[:200]) + "…"
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package meshwantsclient
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
)
|
||||
|
||||
// fakeWants answers the verb revision with its newest trunk revision, and counts each ask.
|
||||
type fakeWants struct {
|
||||
mu sync.Mutex
|
||||
newest int64
|
||||
asks int
|
||||
}
|
||||
|
||||
func (f *fakeWants) call(key string, body any) (json.RawMessage, error) {
|
||||
if key != RevisionVerb {
|
||||
return nil, fmt.Errorf("asked %s", key)
|
||||
}
|
||||
if in := body.(seat.RevisionInput); in != (seat.RevisionInput{}) {
|
||||
return nil, fmt.Errorf("asked %+v", in)
|
||||
}
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.asks++
|
||||
r := seat.AssignmentRevision{Revision: f.newest, Kind: kindTrunk, Placements: "p",
|
||||
Catalogs: []seat.CatalogOf{{Branch: "main", Revision: 10 + f.newest}},
|
||||
Entries: []seat.NodeEntryRef{{Node: "anchor", Module: "nats", Digest: fmt.Sprintf("d%d", f.newest)}}}
|
||||
return json.Marshal(seat.RevisionAnswer{Revision: r})
|
||||
}
|
||||
|
||||
func (f *fakeWants) set(newest int64) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.newest = newest
|
||||
}
|
||||
|
||||
func (f *fakeWants) asked() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.asks
|
||||
}
|
||||
|
||||
func revisionEvent(rev int64, kind string) []byte {
|
||||
b, _ := json.Marshal(seat.AssignmentRevisionEvent{V: 1, Revision: rev, Kind: kind})
|
||||
return b
|
||||
}
|
||||
|
||||
// CT10 (design 47 G §245): a part takes a first view of the assignment half by the verb revision, then follows its
|
||||
// event assignment-revision: a newer trunk revision is waited for, said in the status, and held; a candidate's, or one
|
||||
// held, asks nothing; what is no such event is refused.
|
||||
func TestTheAssignmentHalfIsAFirstViewThenItsEvents(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
held := make(chan seat.AssignmentRevision, 8)
|
||||
next := func() seat.AssignmentRevision {
|
||||
t.Helper()
|
||||
select {
|
||||
case r := <-held:
|
||||
return r
|
||||
case <-ctx.Done():
|
||||
t.Fatal("no revision held")
|
||||
return seat.AssignmentRevision{}
|
||||
}
|
||||
}
|
||||
f := &fakeWants{newest: 3}
|
||||
a := NewAssignment(f.call)
|
||||
a.Follower.Wait = time.Nanosecond
|
||||
a.Held = func(r seat.AssignmentRevision) { held <- r }
|
||||
if _, ok := a.Revision(); ok {
|
||||
t.Fatal("a revision before its first view")
|
||||
}
|
||||
if err := a.Start(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r := next(); r.Revision != 3 || len(r.Catalogs) != 1 || r.Catalogs[0].Revision != 13 {
|
||||
t.Fatalf("first view holds %+v", r)
|
||||
}
|
||||
if v, ok := a.Follower.Get("anchor.nats"); !ok || string(v) == "" {
|
||||
t.Errorf("the picture lacks anchor.nats: %s", v)
|
||||
}
|
||||
|
||||
for _, ev := range [][]byte{revisionEvent(3, kindTrunk), revisionEvent(2, kindTrunk), revisionEvent(9, "candidate")} {
|
||||
if err := a.Hear(ctx, ev); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if n := f.asked(); n != 1 {
|
||||
t.Errorf("a revision held, an older one or a candidate's asked again: %d asks", n)
|
||||
}
|
||||
|
||||
f.set(4)
|
||||
if err := a.Hear(ctx, revisionEvent(4, kindTrunk)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r := next(); r.Revision != 4 {
|
||||
t.Fatalf("after assignment-revision 4 the view holds %d", r.Revision)
|
||||
}
|
||||
if s := a.Follower.Status(); s.View != "4" || s.WaitsFor != "" {
|
||||
t.Errorf("status view %q, waits for %q", s.View, s.WaitsFor)
|
||||
}
|
||||
if r, ok := a.Revision(); !ok || r.Revision != 4 {
|
||||
t.Errorf("Revision = %d, %v", r.Revision, ok)
|
||||
}
|
||||
|
||||
for _, data := range []string{`{}`, `not json`, `{"kind":"trunk"}`} {
|
||||
if err := a.Hear(ctx, []byte(data)); err == nil {
|
||||
t.Errorf("Hear(%s) took it", data)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -6,3 +6,12 @@ require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/nats-io/nats.go v1.54.0 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
)
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
|
||||
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
|
||||
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
@@ -79,7 +79,7 @@ var readFamilies = []string{buckets.ComposedRead}
|
||||
var ownBuckets = []string{buckets.Asks, buckets.Differences,
|
||||
buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket,
|
||||
plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket,
|
||||
engineReadsBucket, retiredBucket, copiesBucket}
|
||||
engineReadsBucket, retiredBucket, copiesBucket, writtenBucket}
|
||||
|
||||
// engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key `<node>`, as
|
||||
// the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again.
|
||||
@@ -88,6 +88,9 @@ const engineReadsState = "engine-reads"
|
||||
// engineReadsBucket is that state's bucket.
|
||||
const engineReadsBucket = assignment.Module + "_" + engineReadsState
|
||||
|
||||
// writtenBucket is the bucket of the state written (pipelines.go): the pipeline ends the wants follow.
|
||||
const writtenBucket = assignment.Module + "_" + writtenState
|
||||
|
||||
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
|
||||
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
|
||||
var seatBuckets = []string{plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket,
|
||||
@@ -115,6 +118,7 @@ var keptStates = []announce.State{
|
||||
{Name: engineReadsState, Readable: true},
|
||||
{Name: retiredState}, // read only by the copy into assignments (migrate.go)
|
||||
{Name: copiesState, Readable: true},
|
||||
{Name: writtenState, Readable: true},
|
||||
}
|
||||
|
||||
// kvSet is every bucket the assigner opened, by name.
|
||||
@@ -197,7 +201,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
func startsWithout(name string) bool {
|
||||
switch name {
|
||||
case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket,
|
||||
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket:
|
||||
buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket:
|
||||
return true
|
||||
}
|
||||
return slices.Contains(seatBuckets, name) || name == buckets.Grants
|
||||
@@ -297,6 +301,7 @@ var watched = []string{
|
||||
buckets.Source,
|
||||
buckets.Merges,
|
||||
engineReadsBucket,
|
||||
writtenBucket,
|
||||
}
|
||||
|
||||
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
||||
@@ -485,6 +490,18 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
logger.Printf("%s is not heard, so an ask that ends at the operator channel without an answer is not asked again until the assigner starts again: %v", askEndedSubject, err)
|
||||
}
|
||||
readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], staleOfWorld(w), bus, callWait{heard: func() bool { return w.seatHeard(oc.Seat) }, findings: findings}, logger.Printf)
|
||||
if kv := kvs[writtenBucket]; kv != nil {
|
||||
// The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow
|
||||
// once the first registrations are written (§95, §353).
|
||||
p.ends = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf)
|
||||
err := hearEnds(hub, p.ends)
|
||||
if err == nil && bus != nil {
|
||||
err = followRegistrations(ctx, hub, bus.Call, p.ends)
|
||||
}
|
||||
if err != nil {
|
||||
logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err)
|
||||
}
|
||||
}
|
||||
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
|
||||
}
|
||||
if bus != nil && kvs[buckets.Puts] != nil {
|
||||
|
||||
@@ -58,6 +58,8 @@ type passer struct {
|
||||
// neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the
|
||||
// first pass after a start, which clears what a previous process may have raised (startClears).
|
||||
neither map[string]neitherOne
|
||||
// ends are the pipeline ends the wants follow (pipelines.go), once the first registrations are heard; until then, and nil, the trunk-only want.
|
||||
ends *pipelineEnds
|
||||
// acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket).
|
||||
acts *putActs
|
||||
// noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is
|
||||
@@ -267,9 +269,16 @@ func (p *passer) wants(ctx context.Context) {
|
||||
moduleGraph := p.moduleGraph()
|
||||
served := p.served(ctx, get, moduleGraph, nodes)
|
||||
offering := p.offering(ctx, get, manifests, nodes) // bus state (ADR 0614 §5)
|
||||
assignedBranches := p.w.placedBranches()
|
||||
for name := range names {
|
||||
fp := trunk[name].Fp2
|
||||
name := name
|
||||
// A want follows a passed pipeline: a module whose fingerprint no pipeline passed at on every branch its placements
|
||||
// chose, and whose wants do not name it already, gets none (design 47 G §29, §35).
|
||||
branches := p.ends.branchesOf(name, assignedBranches)
|
||||
if fp != "" && !p.ends.mayWant(name, fp, branches, wanted[name]) {
|
||||
continue
|
||||
}
|
||||
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp),
|
||||
Self: func(node string) bool { return placed[node] },
|
||||
Composed: func(node, fp string) (assign.ComposedDigest, bool) {
|
||||
@@ -359,6 +368,9 @@ func (p *passer) wants(ctx context.Context) {
|
||||
want := want
|
||||
p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeWant(want) })
|
||||
}
|
||||
if len(out.Wants) > 0 {
|
||||
p.ends.wrote(name, fp, branches)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1128,10 +1128,10 @@ func aReleasesApproveAnswersTheMoveOfItsBuildAlone(t *testing.T) {
|
||||
p.logf = t.Logf
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: commit, Src: 7, FpSrc: c.fpSrc, Path: meshcatalogclient.RootPath})
|
||||
p.w.put(buckets.Modules, "web", mod)
|
||||
src, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}})
|
||||
p.w.putEntry(buckets.Source, mdc.SourceMainKey, src, 7, time.Now())
|
||||
later, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}})
|
||||
p.w.putEntry(buckets.Source, mdc.SourceMainKey, later, 8, time.Now()) // a later merge leaves revision 7's release
|
||||
src, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}})
|
||||
p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), src, 7, time.Now())
|
||||
later, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}})
|
||||
p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), later, 8, time.Now()) // a later merge leaves revision 7's release
|
||||
if c.mergedAt != "" {
|
||||
merge, err := mdc.EncodeMerge(mdseat.MergesByGroupValue{Group: "feat/web", Ask: "merge.feat_web.0123456789ab", Pulls: []mdseat.MergesByGroupValuePullsItem{
|
||||
{Repository: "novox/web", Number: 3, Head: strings.Repeat("3", 40), State: string(mdc.MergeMerged), Commit: c.mergedAt}}})
|
||||
|
||||
@@ -0,0 +1,415 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient"
|
||||
mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// The wants follow a passed pipeline (design 47 G §29, §35, §325, §352): the runner says each pipeline's end as
|
||||
// pipeline-ended, and a group's as group-ended. The assigner keeps, in its old state `written`, one key per module and
|
||||
// watched branch: the source revision and end count of the newest end it heard for it, and whether the wants were written
|
||||
// from it. A newer end replaces it by compare-and-set; an older one, or the same one again, writes nothing.
|
||||
|
||||
// writtenState is the assigner's state of the pipeline ends its wants follow; writtenTool reads it.
|
||||
const (
|
||||
writtenState = "written"
|
||||
writtenTool = "assigner_written"
|
||||
)
|
||||
|
||||
// pipelineEndedSubject and groupEndedSubject are the runner's events the assigner consumes.
|
||||
const (
|
||||
pipelineEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventPipelineEnded + ".>"
|
||||
groupEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventGroupEnded + ".>"
|
||||
)
|
||||
|
||||
// End is one value of the state written: the newest end of a module's pipeline on a watched branch.
|
||||
type End struct {
|
||||
Source int64 `json:"source"`
|
||||
EndCount int64 `json:"end-count"`
|
||||
Fp string `json:"fp"`
|
||||
Commit string `json:"commit,omitempty"`
|
||||
// Passed: the pipeline ended passed, every required step passed (and, for a group, the group ended passed).
|
||||
Passed bool `json:"passed"`
|
||||
// Written: the wants were written from this end.
|
||||
Written bool `json:"written,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
// endKey is the key of a module on a watched branch, each name escaped as a key's name is.
|
||||
func endKey(module, branch string) string {
|
||||
return keys.EscapeName(module) + "." + keys.EscapeName(branch)
|
||||
}
|
||||
|
||||
// newer says whether an end of (source, count) replaces the one kept: a later source revision, or the same one ended again.
|
||||
func (e End) newer(source, count int64) bool {
|
||||
return source > e.Source || (source == e.Source && count > e.EndCount)
|
||||
}
|
||||
|
||||
// requiredPassed says whether no required step of a pipeline stands failed, unasked or unended (design 47 G §30: a failed
|
||||
// optional step holds nothing).
|
||||
func requiredPassed(p mpr.PipelineEnded) bool {
|
||||
for _, s := range p.Steps {
|
||||
if s.Status != mpr.StepResultStatusRequired {
|
||||
continue
|
||||
}
|
||||
switch s.Outcome {
|
||||
case mpr.StepResultOutcomeFailed, mpr.StepResultOutcomeCouldNotRun, mpr.StepResultOutcomeDeclined, mpr.StepResultOutcomeNotAsked,
|
||||
mpr.StepResultOutcomeCancelled, mpr.StepResultOutcomeSuperseded, mpr.StepResultOutcomeDidNotBuild,
|
||||
mpr.StepResultOutcomeRunning, mpr.StepResultOutcomeWaiting, mpr.StepResultOutcomeWaitingForABuildNode:
|
||||
return false
|
||||
}
|
||||
}
|
||||
return p.Outcome == mpr.PipelineEndedOutcomePassed
|
||||
}
|
||||
|
||||
// endStore is the kept state written: the bus bucket, or a test's.
|
||||
type endStore interface {
|
||||
// get is a key's value and revision; none for a key not there.
|
||||
get(key string) (value []byte, revision uint64, err error)
|
||||
// swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not.
|
||||
swap(key string, value []byte, revision uint64) error
|
||||
// keys are the keys that stand, and drop deletes one.
|
||||
keys() ([]string, error)
|
||||
drop(key string) error
|
||||
}
|
||||
|
||||
var errSwapped = errors.New("the key changed")
|
||||
|
||||
// kvEnds is endStore over a bus bucket.
|
||||
type kvEnds struct {
|
||||
ctx context.Context
|
||||
kv jetstream.KeyValue
|
||||
}
|
||||
|
||||
func (k kvEnds) get(key string) ([]byte, uint64, error) {
|
||||
e, err := k.kv.Get(k.ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return nil, 0, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return e.Value(), e.Revision(), nil
|
||||
}
|
||||
|
||||
func (k kvEnds) swap(key string, value []byte, revision uint64) error {
|
||||
var err error
|
||||
if revision == 0 {
|
||||
_, err = k.kv.Create(k.ctx, key, value)
|
||||
} else {
|
||||
_, err = k.kv.Update(k.ctx, key, value, revision)
|
||||
}
|
||||
if errors.Is(err, jetstream.ErrKeyExists) || (err != nil && isWrongSequence(err)) {
|
||||
return errSwapped
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (k kvEnds) keys() ([]string, error) {
|
||||
ks, err := k.kv.Keys(k.ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil, nil
|
||||
}
|
||||
return ks, err
|
||||
}
|
||||
|
||||
func (k kvEnds) drop(key string) error { return k.kv.Delete(k.ctx, key) }
|
||||
|
||||
func isWrongSequence(err error) bool {
|
||||
var ae *jetstream.APIError
|
||||
return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
|
||||
}
|
||||
|
||||
// pipelineEnds follows the runner's ends. A nil *pipelineEnds is the trunk-only want, kept until the first registrations
|
||||
// (design 47 G §95, §353).
|
||||
type pipelineEnds struct {
|
||||
st endStore
|
||||
wake func()
|
||||
logf func(string, ...any)
|
||||
// on is whether the wants follow the ends: set by the first registration heard, never before (design 47 G §95, §353).
|
||||
on atomic.Bool
|
||||
mu sync.Mutex
|
||||
// firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design
|
||||
// 47 G §14).
|
||||
firsts map[string]string
|
||||
}
|
||||
|
||||
func newPipelineEnds(st endStore, wake func(), logf func(string, ...any)) *pipelineEnds {
|
||||
return &pipelineEnds{st: st, wake: wake, logf: logf, firsts: map[string]string{}}
|
||||
}
|
||||
|
||||
// take keeps one pipeline's end if it is newer than the one kept for its module and branch, by compare-and-set.
|
||||
func (e *pipelineEnds) take(p mpr.PipelineEnded, passed bool) error {
|
||||
key := endKey(p.Module, p.Branch)
|
||||
for {
|
||||
raw, rev, err := e.st.get(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if raw != nil {
|
||||
var kept End
|
||||
if err := json.Unmarshal(raw, &kept); err == nil && !kept.newer(p.Source, p.EndCount) {
|
||||
return nil // an older end, or this one again
|
||||
}
|
||||
}
|
||||
end := End{Source: p.Source, EndCount: p.EndCount, Fp: p.Fp, Commit: p.Commit, Passed: passed, Why: p.Why}
|
||||
value, err := json.Marshal(end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) {
|
||||
continue
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
e.wake()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// markWritten says the wants of module on branch were written from the end it keeps, if that is still the end of fp.
|
||||
func (e *pipelineEnds) markWritten(module, branch, fp string) {
|
||||
key := endKey(module, branch)
|
||||
for {
|
||||
raw, rev, err := e.st.get(key)
|
||||
var kept End
|
||||
if err != nil || raw == nil || json.Unmarshal(raw, &kept) != nil || kept.Fp != fp || kept.Written {
|
||||
return
|
||||
}
|
||||
kept.Written = true
|
||||
value, _ := json.Marshal(kept)
|
||||
if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) {
|
||||
continue
|
||||
} else if err != nil {
|
||||
e.logf("the end of %s on %s is not marked written: %v", module, branch, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// passedAt says whether the newest end kept for module on branch passed at fp.
|
||||
func (e *pipelineEnds) passedAt(module, branch, fp string) bool {
|
||||
raw, _, err := e.st.get(endKey(module, branch))
|
||||
var kept End
|
||||
return err == nil && raw != nil && json.Unmarshal(raw, &kept) == nil && kept.Passed && kept.Fp == fp
|
||||
}
|
||||
|
||||
// branchOf is the watched branch a placement names, else the first of the module's registration, else the trunk's.
|
||||
func (e *pipelineEnds) branchOf(module, placed string) string {
|
||||
if placed != "" {
|
||||
return placed
|
||||
}
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if b := e.firsts[module]; b != "" {
|
||||
return b
|
||||
}
|
||||
return trunkBranch
|
||||
}
|
||||
|
||||
// hearPipeline takes a pipeline-ended. A pipeline of a group writes nothing on its own end, and a run on request delivers
|
||||
// nothing (design 47 G §29, §64).
|
||||
func (e *pipelineEnds) hearPipeline(m jetstream.Msg) {
|
||||
p, err := mprc.PipelineEnded(m.Data())
|
||||
switch {
|
||||
case err != nil:
|
||||
e.logf("%s is not taken: %v", m.Subject(), err)
|
||||
case p.Group != "" || p.Run != "":
|
||||
default:
|
||||
if err := e.take(p, requiredPassed(p)); err != nil {
|
||||
e.logf("%s: %v; it is heard again", m.Subject(), err)
|
||||
_ = m.Nak()
|
||||
return
|
||||
}
|
||||
}
|
||||
_ = m.Ack()
|
||||
}
|
||||
|
||||
// hearGroup takes a group-ended: each member's end, passed only if the group passed.
|
||||
func (e *pipelineEnds) hearGroup(m jetstream.Msg) {
|
||||
g, err := mprc.GroupEnded(m.Data())
|
||||
if err != nil {
|
||||
e.logf("%s is not taken: %v", m.Subject(), err)
|
||||
_ = m.Ack()
|
||||
return
|
||||
}
|
||||
for _, p := range g.Pipelines {
|
||||
if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil {
|
||||
e.logf("%s: %v; it is heard again", m.Subject(), err)
|
||||
_ = m.Nak()
|
||||
return
|
||||
}
|
||||
}
|
||||
_ = m.Ack()
|
||||
}
|
||||
|
||||
// hearEnds registers the two events on the assigner's hub.
|
||||
func hearEnds(hub *process.EventHub, e *pipelineEnds) error {
|
||||
if err := hub.Handle(pipelineEndedSubject, e.hearPipeline); err != nil {
|
||||
return fmt.Errorf("%s: %w", pipelineEndedSubject, err)
|
||||
}
|
||||
if err := hub.Handle(groupEndedSubject, e.hearGroup); err != nil {
|
||||
return fmt.Errorf("%s: %w", groupEndedSubject, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// writtenTools is the tool assigner_written: every key of the state written, with its end.
|
||||
func writtenTools(st store) []stdio.Tool {
|
||||
return []stdio.Tool{{Name: writtenTool, Description: "the pipeline ends the assigner's wants follow: per module and watched branch, the source revision and end count of the newest end heard, whether it passed and whether the wants were written from it.",
|
||||
Run: func(map[string]any) (any, error) {
|
||||
ks, err := st.keys(writtenState)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sort.Strings(ks)
|
||||
out := map[string]End{}
|
||||
for _, k := range ks {
|
||||
raw, _, ok, err := st.get(writtenState, k)
|
||||
if err != nil || !ok {
|
||||
continue
|
||||
}
|
||||
var e End
|
||||
if json.Unmarshal(raw, &e) == nil {
|
||||
out[k] = e
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}}}
|
||||
}
|
||||
|
||||
// branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one.
|
||||
func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string {
|
||||
if !e.following() {
|
||||
return nil
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
var out []string
|
||||
for _, b := range placed[module] {
|
||||
b = e.branchOf(module, b)
|
||||
if !seen[b] {
|
||||
seen[b] = true
|
||||
out = append(out, b)
|
||||
}
|
||||
}
|
||||
if len(out) == 0 {
|
||||
out = append(out, e.branchOf(module, ""))
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already
|
||||
// (a node placed later, a module taken away).
|
||||
func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool {
|
||||
if !e.following() {
|
||||
return true
|
||||
}
|
||||
for _, w := range wanted {
|
||||
if w.Fp == fp {
|
||||
return true
|
||||
}
|
||||
}
|
||||
for _, b := range branches {
|
||||
if !e.passedAt(module, b, fp) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// wrote marks the end the wants of module were written from.
|
||||
func (e *pipelineEnds) wrote(module, fp string, branches []string) {
|
||||
if !e.following() {
|
||||
return
|
||||
}
|
||||
for _, b := range branches {
|
||||
e.markWritten(module, b, fp)
|
||||
}
|
||||
}
|
||||
|
||||
// following says whether the wants follow the pipeline ends yet; a nil follower never does.
|
||||
func (e *pipelineEnds) following() bool { return e != nil && e.on.Load() }
|
||||
|
||||
// registered takes a module's registration (design 47 G §353): the first one heard switches the wants to the ends, its first
|
||||
// watched branch is the default a placement runs, and the ends kept for a branch it no longer watches are deleted (§352). gone
|
||||
// is a registration removed: all its ends go.
|
||||
func (e *pipelineEnds) registered(reg mpcat.Registration, gone bool) {
|
||||
e.mu.Lock()
|
||||
if gone || len(reg.Branches) == 0 {
|
||||
delete(e.firsts, reg.Module)
|
||||
} else {
|
||||
e.firsts[reg.Module] = reg.Branches[0]
|
||||
}
|
||||
e.mu.Unlock()
|
||||
kept, err := e.st.keys()
|
||||
if err != nil {
|
||||
e.logf("the ends kept cannot be listed: %v", err)
|
||||
}
|
||||
prefix := keys.EscapeName(reg.Module) + "."
|
||||
for _, k := range kept {
|
||||
rest, ok := strings.CutPrefix(k, prefix)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
branch, err := keys.UnescapeName(rest)
|
||||
if err == nil && !gone && slices.Contains(reg.Branches, branch) {
|
||||
continue
|
||||
}
|
||||
if err := e.st.drop(k); err != nil {
|
||||
e.logf("the end %s is not deleted: %v", k, err)
|
||||
}
|
||||
}
|
||||
if !e.on.Swap(true) {
|
||||
e.logf("the first registration is heard (%s): the wants follow the pipelines' ends from now", reg.Module)
|
||||
}
|
||||
e.wake()
|
||||
}
|
||||
|
||||
// followRegistrations follows the registrations through the seat mesh-catalog (design 47 G §353): a first view of its verb
|
||||
// and each registration-changed on the hub. It returns when the hub's handler is registered; the first view is asked on its own
|
||||
// goroutine, the wants keep the trunk-only form until a registration is heard.
|
||||
func followRegistrations(ctx context.Context, hub *process.EventHub, call mcc.Call, e *pipelineEnds) error {
|
||||
regs := mcc.NewRegistrations(call)
|
||||
regs.Follower.Logf = e.logf
|
||||
regs.Follower.Changed = func(module string, value json.RawMessage, gone bool, _ time.Time) {
|
||||
reg := mpcat.Registration{Module: module}
|
||||
if !gone {
|
||||
_ = json.Unmarshal(value, ®)
|
||||
}
|
||||
e.registered(reg, gone)
|
||||
}
|
||||
if err := hub.Handle(mpcat.SubjectRegistrationChanged, func(m jetstream.Msg) {
|
||||
if err := regs.Hear(ctx, m.Data()); err != nil {
|
||||
e.logf("%s: %v", m.Subject(), err)
|
||||
}
|
||||
_ = m.Ack()
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
if err := regs.Start(ctx); err != nil && ctx.Err() == nil {
|
||||
e.logf("the first view of the registrations is not answered: %v", err)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
)
|
||||
|
||||
// memEnds is endStore over a map, with revisions.
|
||||
type memEnds struct {
|
||||
vals map[string][]byte
|
||||
revs map[string]uint64
|
||||
}
|
||||
|
||||
func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} }
|
||||
|
||||
func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil }
|
||||
func (m *memEnds) keys() (out []string, _ error) {
|
||||
for k := range m.vals {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (m *memEnds) drop(key string) error { delete(m.vals, key); delete(m.revs, key); return nil }
|
||||
func (m *memEnds) swap(key string, v []byte, rev uint64) error {
|
||||
if m.revs[key] != rev {
|
||||
return errSwapped
|
||||
}
|
||||
m.vals[key], m.revs[key] = v, rev+1
|
||||
return nil
|
||||
}
|
||||
|
||||
func pipe(module, branch string, source, count int64, fp, outcome string, steps ...mpr.StepResult) mpr.PipelineEnded {
|
||||
return mpr.PipelineEnded{Module: module, Branch: branch, Source: source, EndCount: count, Fp: fp, Outcome: outcome, Steps: steps,
|
||||
Pipeline: branch + "." + module, Commit: "c"}
|
||||
}
|
||||
|
||||
func step(status, outcome string) mpr.StepResult {
|
||||
return mpr.StepResult{Step: "unit", Status: status, Outcome: outcome}
|
||||
}
|
||||
|
||||
// Design 47 G §29, §35, §325, §352 (tests 66, 67, 89): a want follows a passed pipeline, by compare-and-set per module and branch.
|
||||
func TestAWantFollowsAPassedPipeline(t *testing.T) {
|
||||
const fp1, fp2 = "fp1", "fp2"
|
||||
pass, fail := mpr.PipelineEndedOutcomePassed, mpr.PipelineEndedOutcomeFailed
|
||||
req, opt := mpr.StepResultStatusRequired, mpr.StepResultStatusOptional
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
ends []mpr.PipelineEnded
|
||||
group *mpr.GroupEnded
|
||||
module string
|
||||
branch string
|
||||
fp string
|
||||
may bool
|
||||
}{
|
||||
{"a failed required step writes no want", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed"))}, nil, "web", "main", fp1, false},
|
||||
{"a pipeline passed writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true},
|
||||
{"a failed optional step holds nothing", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"), step(opt, "failed"))}, nil, "web", "main", fp1, true},
|
||||
{"a pipeline passed at another fingerprint", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp2, false},
|
||||
{"an older revision ending after a newer one writes nothing", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp1, false},
|
||||
{"another branch writes its own", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "beta", 5, 1, fp1, pass)}, nil, "web", "beta", fp1, true},
|
||||
{"a later end of the same revision writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed")), pipe("web", "main", 5, 2, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true},
|
||||
{"the same end again after a later one keeps the later", []mpr.PipelineEnded{pipe("web", "main", 5, 2, fp1, pass), pipe("web", "main", 5, 1, fp1, fail)}, nil, "web", "main", fp1, true},
|
||||
{"a member's own end writes nothing", []mpr.PipelineEnded{func() mpr.PipelineEnded { p := pipe("web", "main", 5, 1, fp1, pass); p.Group = "g"; return p }()}, nil, "web", "main", fp1, false},
|
||||
{"a group ended passed writes its members", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomePassed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, true},
|
||||
{"a group ended failed writes none", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomeFailed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, false},
|
||||
} {
|
||||
woken := 0
|
||||
e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf)
|
||||
e.on.Store(true)
|
||||
for _, p := range c.ends {
|
||||
if p.Group != "" {
|
||||
continue // heard and ignored, as hearPipeline does
|
||||
}
|
||||
if err := e.take(p, requiredPassed(p)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if g := c.group; g != nil {
|
||||
for _, p := range g.Pipelines {
|
||||
if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if got := e.mayWant(c.module, c.fp, []string{c.branch}, nil); got != c.may {
|
||||
t.Errorf("%s: may want = %v, want %v", c.name, got, c.may)
|
||||
}
|
||||
}
|
||||
// A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it).
|
||||
e := newPipelineEnds(newMemEnds(), func() {}, t.Logf)
|
||||
e.on.Store(true)
|
||||
if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) {
|
||||
t.Error("a want naming the fingerprint was held")
|
||||
}
|
||||
// markWritten keeps the end and says it was written from; a newer end replaces it unwritten.
|
||||
st := newMemEnds()
|
||||
e = newPipelineEnds(st, func() {}, t.Logf)
|
||||
e.on.Store(true)
|
||||
_ = e.take(pipe("web", "main", 5, 1, fp1, pass), true)
|
||||
e.wrote("web", fp1, []string{"main"})
|
||||
var end End
|
||||
_ = json.Unmarshal(st.vals[endKey("web", "main")], &end)
|
||||
if !end.Written || end.Source != 5 {
|
||||
t.Errorf("end = %+v, want written at source 5", end)
|
||||
}
|
||||
_ = e.take(pipe("web", "main", 6, 1, fp2, pass), true)
|
||||
end = End{}
|
||||
_ = json.Unmarshal(st.vals[endKey("web", "main")], &end)
|
||||
if end.Written || end.Source != 6 {
|
||||
t.Errorf("a newer end = %+v, want unwritten at source 6", end)
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §29: the wants of a pass follow a pipeline: with the follower on, a module whose fingerprint no pipeline passed at
|
||||
// gets no want and one a pipeline passed at gets it; with it off (nil, the trunk-only want) the pass is as before.
|
||||
func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
|
||||
fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
follow bool
|
||||
passed bool
|
||||
want bool
|
||||
}{{"the trunk-only want", false, false, true}, {"no pipeline passed", true, false, false}, {"a pipeline passed", true, true, true}} {
|
||||
wantsKV := &memKV{values: map[string][]byte{}}
|
||||
p := newPasser(kvSet{buckets.Wants: wantsKV, buckets.Asks: &memKV{values: map[string][]byte{}}}, time.Now())
|
||||
p.say = func(oc.Condition) error { return nil }
|
||||
mdigest := strings.Repeat("e", 64)
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
p.w.put(buckets.Modules, "web", mod)
|
||||
p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}}
|
||||
bkey, _ := builds.BuilderBuildKey("web", fp)
|
||||
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest})
|
||||
p.w.put(buckets.Builds, bkey, built)
|
||||
gkey, _ := grantsstate.PartGrantKey("web", fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
p.w.put(buckets.Grants, gkey, grant)
|
||||
digest := "sha256:" + strings.Repeat("c", 64)
|
||||
have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
p.w.put(buckets.Have, "g14.web", have)
|
||||
composes(t, p.w, "g14", digest)
|
||||
placeOn(t, p.w, "g14", "web")
|
||||
on, _ := presenceValue(true)
|
||||
p.w.put(buckets.Presence, "g14", on)
|
||||
if c.follow {
|
||||
p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf)
|
||||
p.ends.on.Store(true)
|
||||
if c.passed {
|
||||
_ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true)
|
||||
}
|
||||
}
|
||||
p.wants(context.Background())
|
||||
key, _ := mwc.WantKey("g14", "web")
|
||||
if _, got := wantsKV.values[key]; got != c.want {
|
||||
t.Errorf("%s: a want written = %v, want %v", c.name, got, c.want)
|
||||
}
|
||||
if c.follow && c.passed {
|
||||
raw, _, _ := p.ends.st.get(endKey("web", trunkBranch))
|
||||
var end End
|
||||
_ = json.Unmarshal(raw, &end)
|
||||
if !end.Written {
|
||||
t.Errorf("%s: the end was not marked written", c.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §95, §130, §353, §352 (test 100): the wants keep the trunk-only form until the first registration is heard and
|
||||
// switch on it; a branch a registration no longer watches loses its end, and its first branch is a placement's default.
|
||||
func TestTheWantsSwitchOnTheFirstRegistration(t *testing.T) {
|
||||
st := newMemEnds()
|
||||
woken := 0
|
||||
e := newPipelineEnds(st, func() { woken++ }, t.Logf)
|
||||
_ = e.take(pipe("web", "main", 5, 1, "fp1", mpr.PipelineEndedOutcomeFailed), false)
|
||||
_ = e.take(pipe("web", "beta", 5, 1, "fp1", mpr.PipelineEndedOutcomePassed), true)
|
||||
if e.following() || !e.mayWant("web", "fp1", []string{"main"}, nil) {
|
||||
t.Fatal("the follower held a want before any registration")
|
||||
}
|
||||
e.registered(mpcat.Registration{Module: "web", Branches: []string{"beta"}}, false)
|
||||
if !e.following() || woken == 0 {
|
||||
t.Error("the first registration did not switch the wants on")
|
||||
}
|
||||
if e.mayWant("web", "fp1", []string{"main"}, nil) || !e.mayWant("web", "fp1", []string{"beta"}, nil) {
|
||||
t.Error("after the switch the wants did not follow the ends")
|
||||
}
|
||||
if len(st.vals) != 1 || e.branchOf("web", "") != "beta" || e.branchOf("other", "") != trunkBranch {
|
||||
t.Errorf("ends kept %d, default branch %q: an unwatched branch's end stayed, or the default is wrong", len(st.vals), e.branchOf("web", ""))
|
||||
}
|
||||
e.registered(mpcat.Registration{Module: "web"}, true)
|
||||
if len(st.vals) != 0 {
|
||||
t.Errorf("a registration gone left %d ends", len(st.vals))
|
||||
}
|
||||
}
|
||||
@@ -404,6 +404,7 @@ func serveTools() error {
|
||||
tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...)
|
||||
// What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it.
|
||||
tools = append(tools, enginereads.Tool(engineReadsTool, "the assigner", enginereads.KeptState(stdio.State(engineReadsState)), false, time.Now))
|
||||
tools = append(tools, writtenTools(runtimeStore{})...)
|
||||
tools = append(tools, askAgainTools(runtimeStore{}, stdio.Ask, time.Now)...)
|
||||
return stdio.Serve("", tools)
|
||||
}
|
||||
|
||||
@@ -315,6 +315,9 @@ func (s *wantsTools) tools() []stdio.Tool {
|
||||
}
|
||||
var out []stdio.Tool
|
||||
for _, v := range meshwants.Verbs() {
|
||||
if run[v] == nil {
|
||||
continue // a verb of the seat this module does not serve yet: revision, entry, difference (AS11), input (AS5)
|
||||
}
|
||||
out = append(out, stdio.Tool{Name: meshwants.Seat + "." + v, Run: run[v]})
|
||||
}
|
||||
return out
|
||||
|
||||
@@ -86,7 +86,7 @@ func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at t
|
||||
if bucket == plc.AssignmentsBucket {
|
||||
_, _ = w.assignmentKeys.Put(key, value, revision, false) // put took it already; this is its revision
|
||||
}
|
||||
if bucket == buckets.Source && key == mdc.SourceMainKey {
|
||||
if bucket == buckets.Source && key == mdc.SourceKey(trunkBranch) {
|
||||
if s, err := mdc.DecodeSource(value); err == nil && s.Group != "" {
|
||||
w.sourceGroups[revision] = s.Group
|
||||
}
|
||||
@@ -575,3 +575,20 @@ func (w *world) unplacedNodes() []string {
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// trunkBranch is the branch of the trunk-only want, kept until the first registrations (design 47 G §95, §353).
|
||||
const trunkBranch = "main"
|
||||
|
||||
// placedBranches is, by module and node, the watched branch each placement names ("" when it names none).
|
||||
func (w *world) placedBranches() map[string]map[string]string {
|
||||
out := map[string]map[string]string{}
|
||||
for node, a := range w.assignments() {
|
||||
for _, m := range a.Modules {
|
||||
if out[m.Module] == nil {
|
||||
out[m.Module] = map[string]string{}
|
||||
}
|
||||
out[m.Module][node] = m.Branch
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
|
||||
"mesh-delivery-assigner/internal/puts"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -431,7 +432,7 @@ func TestEachRuleHoldsTheWantBack(t *testing.T) {
|
||||
before["g14"] = mwc.Want{Fp: fNew, Composed: dig(fNew)}
|
||||
}
|
||||
out := Wants(m, n, before, nil)
|
||||
if got := out.Wants["g14"]; got != c.want {
|
||||
if got := out.Wants["g14"]; !reflect.DeepEqual(got, c.want) {
|
||||
t.Errorf("%s: g14's want = %+v, want %+v", c.name, got, c.want)
|
||||
}
|
||||
if got := out.Rules["g14"].Rule; got != c.rule {
|
||||
|
||||
@@ -65,7 +65,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
|
||||
"assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1,
|
||||
"engine-reads": 1}; !reflect.DeepEqual(states, want) {
|
||||
"engine-reads": 1, "written": 5}; !reflect.DeepEqual(states, want) {
|
||||
t.Errorf("state = %v, want %v", states, want)
|
||||
}
|
||||
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
|
||||
@@ -96,7 +96,7 @@ func TestManifest(t *testing.T) {
|
||||
// which lists and asks again what each node's engine reads.
|
||||
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants",
|
||||
"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies",
|
||||
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again"}) {
|
||||
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) {
|
||||
t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools)
|
||||
}
|
||||
if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs {
|
||||
@@ -123,11 +123,11 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
|
||||
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "node-engine.manifest-reads-changed"}) {
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) {
|
||||
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes)
|
||||
}
|
||||
if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
"seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||
"seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||
"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
|
||||
"seat:node-backup.backed-up", "seat:node-backup.now",
|
||||
"seat:node-engine.manifest-reads", // ADR 0618 §3
|
||||
|
||||
@@ -39,18 +39,22 @@
|
||||
"operator-channel"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-catalog.registration-changed",
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
"mesh-catalog.seat-definition-removed",
|
||||
"mesh-delivery-indexer.views.graph",
|
||||
"mesh-delivery-indexer.views.manifests",
|
||||
"mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-teller.ask-ended",
|
||||
"mesh-pipeline-runner.pipeline-ended",
|
||||
"mesh-pipeline-runner.group-ended",
|
||||
"node-engine.manifest-reads-changed"
|
||||
],
|
||||
"invokes": [
|
||||
"mesh-delivery-indexer.graph",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"seat:mesh-catalog.registrations",
|
||||
"seat:mesh-catalog.seat-definitions",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-hand-acts.hand-acts",
|
||||
@@ -90,7 +94,8 @@
|
||||
"assigner_engine_reads",
|
||||
"assigner_seat_list",
|
||||
"assigner_seat_list_again",
|
||||
"assigner_ask_again"
|
||||
"assigner_ask_again",
|
||||
"assigner_written"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "assignments", "history": 64 },
|
||||
@@ -108,7 +113,8 @@
|
||||
{ "name": "seat-holders", "history": 5 },
|
||||
{ "name": "view-marks", "history": 1 },
|
||||
{ "name": "first-views", "history": 1 },
|
||||
{ "name": "engine-reads", "history": 1 }
|
||||
{ "name": "engine-reads", "history": 1 },
|
||||
{ "name": "written", "history": 5 }
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
|
||||
@@ -591,6 +591,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := ix.Reconcile(ctx); err != nil {
|
||||
logger.Printf("dropping the modules the trunk does not hold at start: %v", err)
|
||||
}
|
||||
// Every contracts module's trunk definitions read before any manifest is judged and before the forge bridge's stream is
|
||||
// followed (design 47, the indexer reads every definition before it judges): a definition merged while it was down, or
|
||||
// one no event announced, is never missing from the judging below. Once the state is bound, so on its first binding too.
|
||||
// Through the seat git (milestone 47/G W4.15, design 47 *A part that cannot read the forge keeps what it has*): a seat that
|
||||
// does not answer is waited for at most index.GitWaitBound before the judging below runs on the kept definitions, said in
|
||||
// indexer_first_views and as view-waits past five minutes, and read again on the next forge message heard.
|
||||
defs := gitWaits(ctx, ix, findings, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views does not say the wait for the seat git", logger.Printf), logger.Printf)
|
||||
seatDefs.Ready(func(jetstream.KeyValue) { defs.Start(ctx) })
|
||||
// Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578
|
||||
// §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage).
|
||||
verdicts.Ready(func(jetstream.KeyValue) {
|
||||
@@ -603,13 +611,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}()
|
||||
// A pass that fails (a read the bus did not answer) is judged again at the next message or source revision the
|
||||
// indexer hears (index.kickJudge): no timer.
|
||||
j, err := ix.JudgeAtStart(ctx)
|
||||
if err != nil {
|
||||
logger.Printf("judging the manifests at start: %v; judged again at the next message heard", err)
|
||||
judge := func(fromKept bool) {
|
||||
j, err := ix.JudgeAtStart(ctx)
|
||||
if err != nil {
|
||||
logger.Printf("judging the manifests at start (from kept definitions: %v): %v; judged again at the next message heard", fromKept, err)
|
||||
return
|
||||
}
|
||||
logger.Printf("judged %d manifests at start (from kept definitions: %v): %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v",
|
||||
j.Judged, fromKept, j.Written, len(j.Refused), j.Outgrown, j.Waiting)
|
||||
}
|
||||
if !defs.Waiting() {
|
||||
judge(false)
|
||||
return
|
||||
}
|
||||
logger.Printf("judged %d manifests at start: %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v",
|
||||
j.Judged, j.Written, len(j.Refused), j.Outgrown, j.Waiting)
|
||||
// The seat git did not answer: the judging waits for it at most index.GitWaitBound, off the start, so nothing else of
|
||||
// the start is held behind it.
|
||||
go func() { judge(defs.Ready(ctx)) }()
|
||||
})
|
||||
// A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are
|
||||
// made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again
|
||||
@@ -736,6 +753,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
handle(ctx, m, ix, heard, logger)
|
||||
ix.KickFill(ctx)
|
||||
unreadAgain(ctx, ix, logger)
|
||||
defs.Heard(ctx)
|
||||
}, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@@ -180,3 +181,27 @@ func followAssignments(ctx context.Context, hub *process.EventHub, cred process.
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// gitWaits is the indexer's read of the trunk definitions at its start through the seat git (index.DefinitionsAtStart): its
|
||||
// wait said as a first view of the seat's (Owner git, State seat-definitions) in the indexer's first views, and its finding
|
||||
// view-waits written past views.WaitsBound and cleared once the seat answers.
|
||||
func gitWaits(ctx context.Context, ix *index.Indexer, findings, status *process.AwaitedBucket, logf func(string, ...any)) *index.DefinitionsAtStart {
|
||||
finding := func(s views.Status, raised bool) {
|
||||
w := sdkfindings.Writer(ctx, findings.KV)
|
||||
f := views.ViewWaitsFinding(index.Module, s, time.Now())
|
||||
var err error
|
||||
if raised {
|
||||
err = w.Put(f)
|
||||
} else {
|
||||
err = w.Delete(f)
|
||||
}
|
||||
if err != nil {
|
||||
logf("views: the finding view-waits of the seat git is not written: %v", err)
|
||||
}
|
||||
}
|
||||
f := &views.Follower{Owner: git.Seat, Tool: "seat:" + git.Seat + "." + git.VerbTrees, State: index.SeatDefinitionsState, Logf: logf,
|
||||
RaiseWaits: func(s views.Status) { finding(s, true) }, ClearWaits: func(s views.Status) { finding(s, false) }}
|
||||
status.Ready(func(kv jetstream.KeyValue) { f.Said = views.KeepStatus(ctx, kv, logf) })
|
||||
return &index.DefinitionsAtStart{Read: ix.ReadTrunkDefinitions, Waits: f, Logf: logf,
|
||||
Judge: func(ctx context.Context) error { _, err := ix.JudgeAtStart(ctx); return err }}
|
||||
}
|
||||
|
||||
@@ -125,6 +125,36 @@ func TestTheControlRecords(t *testing.T) {
|
||||
// and source, says seat-definition-changed only when a digest moved, keeps the trunk's entry over one that does not
|
||||
// parse (its finding), removes a pull request's entries when it closes, keeps a verdict's out-of-step files, and
|
||||
// raises generated-types-out-of-step from a trunk build end, cleared only by a clean one of the same module.
|
||||
// Milestone 47/G W4.15, design 47 test 54: the seat git not answering at start holds the judging at most its bound, which
|
||||
// then runs on the kept definitions with the wait said; the read once the forge is heard holds the view and judges again.
|
||||
t.Run("W4.15 the start waits for the seat git, judges from kept definitions, and judges again once it answers", func(t *testing.T) {
|
||||
answers := []error{fmt.Errorf("mesh-contracts: %w", errForgeRead), fmt.Errorf("again: %w", errForgeRead), nil}
|
||||
var said []string
|
||||
judged := 0
|
||||
d := &DefinitionsAtStart{Bound: time.Millisecond, Waits: waitsFake{&said},
|
||||
Read: func(context.Context) (int, error) { err := answers[0]; answers = answers[1:]; return 1, err },
|
||||
Judge: func(context.Context) error { judged++; return nil }}
|
||||
ctx := context.Background()
|
||||
d.Start(ctx)
|
||||
if !d.Waiting() || len(said) != 1 || said[0] != "waits "+GitDefinitionsView {
|
||||
t.Fatalf("waiting %v, said %q: want the wait for the seat git said once", d.Waiting(), said)
|
||||
}
|
||||
if !d.Ready(ctx) || judged != 0 {
|
||||
t.Fatalf("the start past its bound: want it judged from the kept definitions, judged again %d", judged)
|
||||
}
|
||||
d.Heard(ctx) // the forge heard, the seat still not answering: the wait stands, said once
|
||||
if !d.Waiting() || len(said) != 1 || judged != 0 {
|
||||
t.Fatalf("a read still not answered: waiting %v, said %q, judged %d", d.Waiting(), said, judged)
|
||||
}
|
||||
d.Heard(ctx)
|
||||
if d.Waiting() || len(said) != 2 || said[1] != "holds "+GitDefinitionsView || judged != 1 || d.Ready(ctx) {
|
||||
t.Errorf("the seat answering: waiting %v, said %q, judged again %d: want the view held and every manifest judged again once", d.Waiting(), said, judged)
|
||||
}
|
||||
d.Heard(ctx) // nothing waits: no read, no judging
|
||||
if judged != 1 || len(answers) != 0 {
|
||||
t.Errorf("a forge message with nothing waiting read or judged again")
|
||||
}
|
||||
})
|
||||
t.Run("seat definitions", func(t *testing.T) {
|
||||
const repo, m1, m2, m3, head = "novox/mesh-contracts", "m1", "m2", "m3", "h1"
|
||||
manifest := `{"module":"mesh-contracts","version":"1.0","build":{"artifacts":[{"name":"go","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}`
|
||||
@@ -315,6 +345,29 @@ func TestTheControlRecords(t *testing.T) {
|
||||
t.Errorf("said %+v: only a trunk's definition is said (ADR 0636 §4)", c)
|
||||
}
|
||||
}
|
||||
// Design 47 (the indexer reads every definition before it judges): at its start the indexer reads every contracts
|
||||
// module's trunk at its repository's head in the source revision, so a definition merged with no event heard is kept
|
||||
// before any manifest is judged; a repository the source names no head of is read at the module's own commit.
|
||||
kept, _ = meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "mesh-contracts", Manifest: manifest})
|
||||
mkey, _ = meshcatalogclient.ManifestsKey(meshcatalogclient.ManifestDigest([]byte(manifest)))
|
||||
ix.Manifests.(*memStore).vals[mkey] = kept
|
||||
mod, _ = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: meshcatalogclient.ManifestDigest([]byte(manifest)), Repository: repo, Commit: m3, Path: meshcatalogclient.RootPath})
|
||||
ix.Modules.(*memStore).vals["mesh-contracts"] = mod
|
||||
files["h3"] = map[string]string{"seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0500", 1)}
|
||||
ix.source = sourceNow{rev: 7, heads: map[string]string{repo: "h3"}}
|
||||
if n, err := ix.ReadTrunkDefinitions(ctx); err != nil || n != 1 {
|
||||
t.Fatalf("read at start = %d, %v; want the one contracts module", n, err)
|
||||
}
|
||||
if e, _ := entry("git.trunk"); e.Commit != "h3" {
|
||||
t.Errorf("git.trunk after the read at start = %+v, want the source's head h3", e)
|
||||
}
|
||||
ix.source = sourceNow{}
|
||||
if _, err := ix.ReadTrunkDefinitions(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if e, _ := entry("git.trunk"); e.Commit != m3 {
|
||||
t.Errorf("git.trunk read with no source head = %+v, want the module's own commit", e)
|
||||
}
|
||||
|
||||
// ADR 0636 §5, D-I3, D-I5: a seat of the mesh's own contracts is taken only from its domain's contracts module; a
|
||||
// move there is a handover in either order of the two trunks; a second definer is refused with a finding naming both.
|
||||
@@ -636,3 +689,9 @@ func (m memSource) Keys() ([]string, error) {
|
||||
}
|
||||
|
||||
func (m memSource) Get(key string) (*views.Entry, bool, error) { e, ok := m[key]; return e, ok, nil }
|
||||
|
||||
// waitsFake records what the indexer says it waits for and holds.
|
||||
type waitsFake struct{ said *[]string }
|
||||
|
||||
func (w waitsFake) WaitFor(_ context.Context, view string) { *w.said = append(*w.said, "waits "+view) }
|
||||
func (w waitsFake) Holds(view string) { *w.said = append(*w.said, "holds "+view) }
|
||||
|
||||
@@ -13,12 +13,14 @@ package index
|
||||
// contracts module.
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"path"
|
||||
"slices"
|
||||
"strconv"
|
||||
@@ -30,6 +32,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/seatdefs"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/trunk"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
@@ -278,6 +281,49 @@ func (ix *Indexer) SeatDefinitionsAgain(ctx context.Context, module string, pull
|
||||
return out
|
||||
}
|
||||
|
||||
// ReadTrunkDefinitions reads, at the start and before any manifest is judged, the seat definitions of every contracts
|
||||
// module the trunk holds, at its repository's head in the source revision (at the module's own commit where the source
|
||||
// names none), and keeps them as a merge does (design 47, *The indexer reads every definition before it judges*): a
|
||||
// definition merged while the indexer was down, or one no event announced, is kept before a manifest claiming its seat
|
||||
// is judged. It answers how many contracts modules were read; one the forge did not answer is in the error, and is read
|
||||
// again at its next merged message or by indexer_seat_definitions_again.
|
||||
func (ix *Indexer) ReadTrunkDefinitions(ctx context.Context) (int, error) {
|
||||
if ix.SeatDefinitions == nil || ix.ReadFile == nil || ix.Modules == nil || ix.Manifests == nil {
|
||||
return 0, nil
|
||||
}
|
||||
held, err := ix.heldModules(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ix.mu.Lock()
|
||||
heads := maps.Clone(ix.source.heads)
|
||||
ix.mu.Unlock()
|
||||
read := 0
|
||||
var errs []error
|
||||
for _, name := range slices.Sorted(maps.Keys(held)) {
|
||||
m := held[name]
|
||||
manifest, _, err := ix.manifestRaw(ctx, m.Manifest)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s's manifest: %w", name, err))
|
||||
continue
|
||||
}
|
||||
dir := cmp.Or(m.Path, meshcatalogclient.RootPath)
|
||||
commit := m.Commit
|
||||
for r, head := range heads {
|
||||
if trunk.SameRepository(r, m.Repository) && head != "" {
|
||||
commit = head
|
||||
}
|
||||
}
|
||||
for _, c := range contractsModules(map[string]string{dir: manifest}) {
|
||||
read++
|
||||
if err := ix.keepTrunk(ctx, m.Repository, commit, c); err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s at %.12s: %w", name, commit, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
return read, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// keepTrunk writes a contracts module's trunk definitions at a merge: each that parses where its digest moved (said),
|
||||
// each the module's index no longer names removed (said with no digest), and the module's unparsed finding written or
|
||||
// cleared. A definition that does not parse leaves the trunk's entry as it stands.
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// GitWaitBound is how long the indexer waits, at its start, for the seat git to answer the read of every contracts module's
|
||||
// trunk definitions before it judges from the definitions it kept (design 47, *A part that cannot read the forge keeps
|
||||
// what it has, and says so*): longer than a forge's restart, short enough that a start is not held behind a forge down
|
||||
// for good.
|
||||
const GitWaitBound = 2 * time.Minute
|
||||
|
||||
// GitDefinitionsView is what the indexer says it waits for in its first views while the seat git does not answer.
|
||||
const GitDefinitionsView = "trunk seat definitions"
|
||||
|
||||
// ViewWaiter is where the indexer says it waits for the seat git and since when, and raises view-waits past its bound
|
||||
// (mesh-sdk views.Follower's WaitFor and Holds).
|
||||
type ViewWaiter interface {
|
||||
WaitFor(ctx context.Context, view string)
|
||||
Holds(view string)
|
||||
}
|
||||
|
||||
// DefinitionsAtStart is the indexer's read of the trunk definitions at its start through the seat git: the first read,
|
||||
// a wait of at most Bound for the seat when it does not answer, the judging from the kept definitions after it, and the
|
||||
// read again once the seat answers, which judges again every manifest judged from kept definitions.
|
||||
type DefinitionsAtStart struct {
|
||||
// Read reads every contracts module's trunk definitions (Indexer.ReadTrunkDefinitions).
|
||||
Read func(ctx context.Context) (int, error)
|
||||
// Judge judges every manifest again, as at the start (Indexer.JudgeAtStart).
|
||||
Judge func(ctx context.Context) error
|
||||
Waits ViewWaiter
|
||||
Bound time.Duration
|
||||
Logf func(string, ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
answered chan struct{} // closed once a read is answered
|
||||
waiting bool // a read was not answered and no read since was
|
||||
fromKept bool // the judging at start ran on kept definitions
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) init() {
|
||||
if d.answered == nil {
|
||||
d.answered = make(chan struct{})
|
||||
}
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) logf(format string, args ...any) {
|
||||
if d.Logf != nil {
|
||||
d.Logf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// Start reads the trunk definitions once; a seat that does not answer is said as a wait (Waits.WaitFor) and read again on
|
||||
// the next sign that the forge runs (Heard). A read answered with other errors (a manifest it cannot read) is answered.
|
||||
func (d *DefinitionsAtStart) Start(ctx context.Context) {
|
||||
d.mu.Lock()
|
||||
d.init()
|
||||
d.mu.Unlock()
|
||||
d.read(ctx, "at start")
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) read(ctx context.Context, when string) bool {
|
||||
n, err := d.Read(ctx)
|
||||
if errors.Is(err, errForgeRead) {
|
||||
d.mu.Lock()
|
||||
first := !d.waiting
|
||||
d.waiting = true
|
||||
d.mu.Unlock()
|
||||
if first {
|
||||
d.logf("the seat git did not answer the read of %d contracts modules' trunk definitions %s: %v; the indexer judges from the definitions it kept after %s, and reads them again once the seat answers", n, when, err, d.bound())
|
||||
if d.Waits != nil {
|
||||
d.Waits.WaitFor(ctx, GitDefinitionsView)
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
if err != nil {
|
||||
d.logf("reading the seat definitions of %d contracts modules' trunks %s: %v; each is read again at its next merged message or by indexer_seat_definitions_again", n, when, err)
|
||||
} else {
|
||||
d.logf("read the seat definitions of %d contracts modules' trunks %s", n, when)
|
||||
}
|
||||
d.mu.Lock()
|
||||
was := d.waiting
|
||||
d.waiting = false
|
||||
select {
|
||||
case <-d.answered:
|
||||
default:
|
||||
close(d.answered)
|
||||
}
|
||||
again := d.fromKept
|
||||
d.fromKept = false
|
||||
d.mu.Unlock()
|
||||
if was && d.Waits != nil {
|
||||
d.Waits.Holds(GitDefinitionsView)
|
||||
}
|
||||
if again && d.Judge != nil {
|
||||
if err := d.Judge(ctx); err != nil {
|
||||
d.logf("judging again the manifests judged from kept definitions: %v; judged again at the next message heard", err)
|
||||
} else {
|
||||
d.logf("judged again every manifest judged from kept definitions, now that the seat git answers")
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) bound() time.Duration {
|
||||
if d.Bound > 0 {
|
||||
return d.Bound
|
||||
}
|
||||
return GitWaitBound
|
||||
}
|
||||
|
||||
// Ready waits, while the seat git does not answer the read at start, until a read is answered, or Bound passed (no timer: a context's bound), and says whether the
|
||||
// judging that follows runs on the definitions the indexer kept; it marks it so, and the read that answers later judges
|
||||
// again (Read's answer).
|
||||
func (d *DefinitionsAtStart) Ready(ctx context.Context) (fromKept bool) {
|
||||
d.mu.Lock()
|
||||
d.init()
|
||||
answered, waiting := d.answered, d.waiting
|
||||
d.mu.Unlock()
|
||||
if !waiting {
|
||||
return false // read, or not read at all (the seat definitions' state not bound yet): nothing to wait for
|
||||
}
|
||||
wait, cancel := context.WithTimeout(ctx, d.bound())
|
||||
defer cancel()
|
||||
select {
|
||||
case <-answered:
|
||||
return false
|
||||
case <-wait.Done():
|
||||
}
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
select {
|
||||
case <-d.answered:
|
||||
return false
|
||||
default:
|
||||
}
|
||||
d.fromKept = true
|
||||
d.logf("the seat git has not answered within %s: the indexer judges from the definitions it kept", d.bound())
|
||||
return true
|
||||
}
|
||||
|
||||
// Heard is a sign that the forge runs (a message the forge bridge relays): while the indexer waits for the seat git, it
|
||||
// reads the trunk definitions again.
|
||||
func (d *DefinitionsAtStart) Heard(ctx context.Context) {
|
||||
d.mu.Lock()
|
||||
waiting := d.waiting
|
||||
d.mu.Unlock()
|
||||
if waiting {
|
||||
d.read(ctx, "once the forge was heard")
|
||||
}
|
||||
}
|
||||
|
||||
// Waiting says the indexer waits for the seat git.
|
||||
func (d *DefinitionsAtStart) Waiting() bool {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.waiting
|
||||
}
|
||||
@@ -324,8 +324,8 @@ func (t *Teller) catchUp(id string) {
|
||||
t.said(h, *w, err)
|
||||
}
|
||||
|
||||
// mergeAct is what the approve option of the gate's ask binds: the ask's verb (meshdeliveryclient.AskVerb: mesh-delivery.merge, or
|
||||
// mesh-delivery.settle for a person's settle ask, novox/hq ADR 0556 clause 5), this ask, the answer, and the heads it covers.
|
||||
// mergeAct is what the approve option of the gate's ask binds: the ask's verb (meshdeliveryclient.AskVerb: mesh-delivery.merge; the settle ask
|
||||
// went with the verb settle, design 47 G §431), this ask, the answer, and the heads it covers.
|
||||
// It is spelled as the gate composes it (mesh-catalog modules/mesh-delivery-gate approval.go mergeAct, from
|
||||
// modules/mesh-delivery approval.go:63-65) and as mesh-controller verifies it (cmd/mesh-controller/moved.go:98-106).
|
||||
func mergeAct(verb, ask, option string, heads map[string]string) occ.Act {
|
||||
|
||||
@@ -412,16 +412,15 @@ func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { //
|
||||
if len(b.conds) != 0 {
|
||||
t.Errorf("conditions said: %+v", b.conds)
|
||||
}
|
||||
// 1280 (ADR 0556 clause 5): a settle ask holds when its options bind mesh-delivery.settle, and not when they bind a merge.
|
||||
for verb, holds := range map[string]bool{"mesh-delivery.settle": true, "mesh-delivery.merge": false} {
|
||||
// design 47 G §431: the settle ask went with the verb settle, so an ask whose options bind it holds no more.
|
||||
for verb, holds := range map[string]bool{"mesh-delivery.settle": false, "mesh-delivery.merge": true} {
|
||||
b := newBed(t)
|
||||
group := meshdeliveryclient.SettleGroup([]string{"novox/mesh-delivery@" + headA[:12]})
|
||||
key, value, ga := gateAskOf(t, group, verb, twoHeads(), twoHeads(), 1)
|
||||
key, value, ga := gateAskOf(t, "g", verb, twoHeads(), twoHeads(), 1)
|
||||
b.gate[key] = value
|
||||
b.t.OnAsk(key, 1, value)
|
||||
b.deliver(t, warrantFor(ga.Ask, "approve"))
|
||||
if _, written := b.answers.written[answerKey(key)]; written != holds {
|
||||
t.Errorf("a settle ask binding %s: answer written %v, want %v", verb, written, holds)
|
||||
t.Errorf("an ask binding %s: answer written %v, want %v", verb, written, holds)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,6 +23,13 @@ type state struct {
|
||||
History int `json:"history,omitempty"`
|
||||
}
|
||||
|
||||
type stream struct {
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
MaxAgeDays int `json:"max-age-days"`
|
||||
MaxMiB int `json:"max-mib"`
|
||||
}
|
||||
|
||||
type artifact struct {
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
@@ -44,6 +51,7 @@ type manifest struct {
|
||||
Invokes []string `json:"invokes"`
|
||||
Consumes []string `json:"consumes"`
|
||||
Emits []string `json:"emits"`
|
||||
Streams []stream `json:"streams"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
SecretsOwner string `json:"secrets-owner"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
@@ -102,7 +110,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
|
||||
if want := []string{"mesh-delivery-assigner.assigner_read_asks", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
if want := []string{"mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
// design 47 G §387: the keeper's ask for the findings is heard, and the finding stream is declared (§383).
|
||||
if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
@@ -122,8 +131,15 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
|
||||
if want := []string{tellerevents.AskEnded}; !reflect.DeepEqual(m.Emits, want) {
|
||||
t.Errorf("emits = %v, want %v", m.Emits, want)
|
||||
}
|
||||
// No streams, bus or seat (ADR 0437 §4).
|
||||
for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
|
||||
// design 47 G §383, §387: the stream <module>-findings, 7 days and 16 MiB; §368: version is MAJOR.MINOR.
|
||||
if want := []stream{{Name: "mesh-delivery-teller-findings", Events: []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}, MaxAgeDays: 7, MaxMiB: 16}}; !reflect.DeepEqual(m.Streams, want) {
|
||||
t.Errorf("streams = %+v, want %+v", m.Streams, want)
|
||||
}
|
||||
if m.Version != "0.1" {
|
||||
t.Errorf("version = %q, want 0.1", m.Version)
|
||||
}
|
||||
// No bus or seat (ADR 0437 §4).
|
||||
for _, k := range []string{"logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the teller has none", k)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"module": "mesh-delivery-teller",
|
||||
"version": "1",
|
||||
"version": "0.1",
|
||||
"slug": "teller",
|
||||
"tools": [
|
||||
"teller_end_ask",
|
||||
@@ -12,8 +12,17 @@
|
||||
"seat:mesh-hand-acts.hand-act"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-conditions.findings-asked",
|
||||
"mesh-delivery-assigner.views.asks"
|
||||
],
|
||||
"streams": [
|
||||
{
|
||||
"name": "mesh-delivery-teller-findings",
|
||||
"events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"],
|
||||
"max-age-days": 7,
|
||||
"max-mib": 16
|
||||
}
|
||||
],
|
||||
"emits": [
|
||||
"ask-ended"
|
||||
],
|
||||
|
||||
@@ -797,7 +797,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
|
||||
want, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64)})
|
||||
failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed})
|
||||
good, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
|
||||
src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
|
||||
for bucket, kv := range map[string][]byte{view.ModulesBucket: mod} {
|
||||
if _, err := kvs[bucket].Put(ctx, "web", kv); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -809,7 +809,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
|
||||
if _, err := kvs[view.HaveBucket].Put(ctx, "ace.web", failed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceMainKey, src)
|
||||
rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceKey("main"), src)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package view
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/trunk"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
@@ -88,3 +89,7 @@ func PackagesKey(module string) (string, error) {
|
||||
}
|
||||
return meshcatalog.PackagesByModuleKey(keys.EscapeName(module)), nil
|
||||
}
|
||||
|
||||
// trunkSourceKey is the key of the gate's state source for the trunk, the watched branch `main` (design 47 G §305: the source
|
||||
// is kept per watched branch, meshdeliveryclient.SourceKey; the view's reads are of the trunk's).
|
||||
var trunkSourceKey = meshdeliveryclient.SourceKey("main")
|
||||
|
||||
@@ -71,7 +71,7 @@ func (m *Model) applyHistory(e Entry) {
|
||||
}
|
||||
m.modHist[e.Key] = h
|
||||
case SourceBucket:
|
||||
if e.Key != meshdeliveryclient.SourceMainKey || e.Deleted || e.Revision == 0 {
|
||||
if e.Key != trunkSourceKey || e.Deleted || e.Revision == 0 {
|
||||
return
|
||||
}
|
||||
src, err := meshdeliveryclient.DecodeSource(e.Value)
|
||||
|
||||
@@ -49,11 +49,11 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
|
||||
return append([]Entry{modIn(fpLater, commit, 12, to, true), modIn(fpLater, commit, 12, to, false)}, extra...)
|
||||
}
|
||||
src := func(rev uint64, webHead string) Entry {
|
||||
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}})
|
||||
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return Entry{Bucket: SourceBucket, Key: meshdeliveryclient.SourceMainKey, Value: v, At: t0, Revision: rev, History: true}
|
||||
return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: rev, History: true}
|
||||
}
|
||||
taken := func(repository, merge string) Entry {
|
||||
key, _ := trunk.RepositoryKey(repository)
|
||||
@@ -72,7 +72,7 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
|
||||
// web changed at A (fpOld) and at C (fpNew); mail first appeared at A. The source named A, B, then C as web's head.
|
||||
world := func(t *testing.T) []Entry {
|
||||
cur, _ := meshdeliveryclient.DecodeSource(src(12, commitC).Value)
|
||||
current := Entry{Bucket: SourceBucket, Key: meshdeliveryclient.SourceMainKey, Revision: 12, At: t0}
|
||||
current := Entry{Bucket: SourceBucket, Key: trunkSourceKey, Revision: 12, At: t0}
|
||||
current.Value, _ = meshdeliveryclient.EncodeSource(cur)
|
||||
return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current,
|
||||
mod("web", fpOld, commitA, 10, true), modIn(fpNew, commitC, 12, in("t1", "mesh-sdk/go"), true), mod("mail", fpMail, commitA, 0, true),
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
@@ -83,7 +82,7 @@ func (r Reader) Tiers(revision uint64) (Tiers, error) {
|
||||
if r.SourceState.GetEntry == nil {
|
||||
return Tiers{}, fmt.Errorf("the gate's source cannot be read here: name the source revision")
|
||||
}
|
||||
_, rev, err := r.SourceState.GetEntry(meshdeliveryclient.SourceMainKey)
|
||||
_, rev, err := r.SourceState.GetEntry(trunkSourceKey)
|
||||
if err != nil {
|
||||
return Tiers{}, fmt.Errorf("the gate's source: %w", err)
|
||||
}
|
||||
|
||||
@@ -621,7 +621,7 @@ func (m *Model) apply(e Entry) {
|
||||
delete(m.taken, e.Key)
|
||||
}
|
||||
case SourceBucket:
|
||||
if e.Key != meshdeliveryclient.SourceMainKey {
|
||||
if e.Key != trunkSourceKey {
|
||||
return
|
||||
}
|
||||
if e.Deleted {
|
||||
|
||||
@@ -61,11 +61,11 @@ func moduleE(t *testing.T, module, fp string) Entry {
|
||||
|
||||
func sourceE(t *testing.T, revision uint64) Entry {
|
||||
t.Helper()
|
||||
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
|
||||
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return Entry{Bucket: SourceBucket, Key: meshdeliveryclient.SourceMainKey, Value: v, At: t0, Revision: revision}
|
||||
return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: revision}
|
||||
}
|
||||
|
||||
// 1906/D7 AC7: every want is a row beside the have of its node, with where it stands.
|
||||
|
||||
@@ -51,6 +51,13 @@ type state struct {
|
||||
History int `json:"history,omitempty"`
|
||||
}
|
||||
|
||||
type stream struct {
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
MaxAgeDays int `json:"max-age-days"`
|
||||
MaxMiB int `json:"max-mib"`
|
||||
}
|
||||
|
||||
type manifest struct {
|
||||
State []state `json:"state"`
|
||||
Module string `json:"module"`
|
||||
@@ -63,6 +70,7 @@ type manifest struct {
|
||||
Uses []string `json:"uses"`
|
||||
Tools []string `json:"tools"`
|
||||
Emits []string `json:"emits"`
|
||||
Streams []stream `json:"streams"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
SecretsOwner string `json:"secrets-owner"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
@@ -117,7 +125,14 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
|
||||
"seat:node-engine.manifest-reads"}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node; 1585/T24: view_tiers asks indexer_pending, a verb edge (ADR 0614 §2)
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
if want := []string{"mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants",
|
||||
// design 47 G §387: the keeper's ask for the findings is heard, the finding stream declared (§383), and §368: version is MAJOR.MINOR.
|
||||
if want := []stream{{Name: "mesh-delivery-view-findings", Events: []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}, MaxAgeDays: 7, MaxMiB: 16}}; !reflect.DeepEqual(m.Streams, want) {
|
||||
t.Errorf("streams = %+v, want %+v", m.Streams, want)
|
||||
}
|
||||
if m.Version != "0.1" {
|
||||
t.Errorf("version = %q, want 0.1", m.Version)
|
||||
}
|
||||
if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants",
|
||||
"mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.taken"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"module": "mesh-delivery-view",
|
||||
"version": "1",
|
||||
"version": "0.1",
|
||||
"moved": true,
|
||||
"slug": "view",
|
||||
"tools": [
|
||||
@@ -42,6 +42,7 @@
|
||||
{ "name": "findings", "history": 1 }
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-conditions.findings-asked",
|
||||
"mesh-delivery-assigner.views.asks",
|
||||
"mesh-delivery-assigner.views.assignments",
|
||||
"mesh-delivery-assigner.views.wants",
|
||||
@@ -49,6 +50,14 @@
|
||||
"mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-indexer.views.taken"
|
||||
],
|
||||
"streams": [
|
||||
{
|
||||
"name": "mesh-delivery-view-findings",
|
||||
"events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"],
|
||||
"max-age-days": 7,
|
||||
"max-mib": 16
|
||||
}
|
||||
],
|
||||
"invokes": [
|
||||
"mesh-delivery-assigner.assigner_read_asks",
|
||||
"mesh-delivery-assigner.assigner_read_differences",
|
||||
|
||||
Reference in New Issue
Block a user