From 18c954ddcd2d0d8167101a0e88e656a875675752 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:45:50 +0100 Subject: [PATCH] CT10: meshwantsclient holds a part's view of the assignment half, a first view then assignment-revision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The typed first view and the following of each half sit in the seats' client packages, built on go/views (design 47 G §245), so no part follows the assignment half by a follower of its own: a newer trunk revision heard is waited for, said in the status, asked and held; a candidate's, or one held, asks nothing. Builds: design 47 G §245, §365 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../go/clients/meshwantsclient/assignment.go | 176 ++++++++++++++++++ .../meshwantsclient/assignment_test.go | 116 ++++++++++++ contracts/go/clients/meshwantsclient/go.mod | 9 + contracts/go/clients/meshwantsclient/go.sum | 24 +++ 4 files changed, 325 insertions(+) create mode 100644 contracts/go/clients/meshwantsclient/assignment.go create mode 100644 contracts/go/clients/meshwantsclient/assignment_test.go create mode 100644 contracts/go/clients/meshwantsclient/go.sum diff --git a/contracts/go/clients/meshwantsclient/assignment.go b/contracts/go/clients/meshwantsclient/assignment.go new file mode 100644 index 00000000..19a716b1 --- /dev/null +++ b/contracts/go/clients/meshwantsclient/assignment.go @@ -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 `_first_views` and `_first_view_again` read and steer it as every first view. +// Its picture is keyed `.`, 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) +} diff --git a/contracts/go/clients/meshwantsclient/assignment_test.go b/contracts/go/clients/meshwantsclient/assignment_test.go new file mode 100644 index 00000000..9d511a02 --- /dev/null +++ b/contracts/go/clients/meshwantsclient/assignment_test.go @@ -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) + } + } +} diff --git a/contracts/go/clients/meshwantsclient/go.mod b/contracts/go/clients/meshwantsclient/go.mod index 470f8994..88f343dc 100644 --- a/contracts/go/clients/meshwantsclient/go.mod +++ b/contracts/go/clients/meshwantsclient/go.mod @@ -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 +) diff --git a/contracts/go/clients/meshwantsclient/go.sum b/contracts/go/clients/meshwantsclient/go.sum new file mode 100644 index 00000000..4ac8a82c --- /dev/null +++ b/contracts/go/clients/meshwantsclient/go.sum @@ -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=