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

2026-11-11 11:53:07 +01:00
parent 3b31e6df05 735724899b
commit 55d8e9e73d
4 changed files with 325 additions and 0 deletions
@@ -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=