novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

AS2: the wants keep the trunk-only form until the first registration, and follow the pipelines' ends from it

Browse Source 
Builds: design 47 G §95, §130, §353, §352

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 11:54:50 +01:00
parent 55d8e9e73d
commit 1778888acc
6 changed files with 143 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+7 -3
View File
@@ -493,9 +493,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
if kv := kvs[writtenBucket]; kv != nil {
// The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow
// once the first registrations are written (§95, §353).
p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf)
if err := hearEnds(hub, p.heard); err != nil {
logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err)
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)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+1 -3
View File
@@ -58,10 +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); nil is the trunk-only want, kept until the first registrations.
// 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
// heard is the same follower while the wants do not follow it yet (ends is nil).
heard *pipelineEnds
// acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket).
acts *putActs
// noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go
+92 -4
View File
@@ -5,13 +5,19 @@ import (
"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"
@@ -81,6 +87,9 @@ type endStore interface {
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")
@@ -115,6 +124,16 @@ func (k kvEnds) swap(key string, value []byte, revision uint64) error {
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
@@ -126,7 +145,9 @@ type pipelineEnds struct {
st endStore
wake func()
logf func(string, ...any)
mu sync.Mutex
// 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
@@ -278,7 +299,7 @@ func writtenTools(st store) []stdio.Tool {
// branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one.
func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string {
if e == nil {
if !e.following() {
return nil
}
seen := map[string]bool{}
@@ -300,7 +321,7 @@ func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]st
// mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already
// (a node placed later, a module taken away).
func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool {
if e == nil {
if !e.following() {
return true
}
for _, w := range wanted {
@@ -318,10 +339,77 @@ func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[
// wrote marks the end the wants of module were written from.
func (e *pipelineEnds) wrote(module, fp string, branches []string) {
if e == nil {
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, &reg)
}
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
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go
+39
View File
@@ -10,6 +10,7 @@ import (
"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"
@@ -34,6 +35,13 @@ type memEnds struct {
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
@@ -79,6 +87,7 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) {
} {
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
@@ -100,12 +109,14 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) {
}
// 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
@@ -153,6 +164,7 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
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)
}
@@ -172,3 +184,30 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
}
}
}
// 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))
}
}
modules/mesh-delivery-assigner/manifest_test.go
+2 -2
View File
@@ -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, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "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
modules/mesh-delivery-assigner/module.json
+2
View File
@@ -39,6 +39,7 @@
"operator-channel"
],
"consumes": [
"mesh-catalog.registration-changed",
"mesh-catalog.seat-definition-changed",
"mesh-catalog.seat-definition-removed",
"mesh-delivery-indexer.views.graph",
@@ -53,6 +54,7 @@
"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",