diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index b0f0d8a..7728190 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -305,6 +305,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + // Its taken state (ADR 0556 clause 1) is bound when it appears, as every state a delivery adds (issue 1790): until then + // a merge's taken key is said and not written, and the rest of the indexer runs. + taken := process.AwaitBucket(ctx, js, conn, delivery.TakenBucket, "no repository's taken key is written", logger.Printf) // Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a // catalogue merge's placement is said and not written, and the rest of the indexer runs. placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf) @@ -341,7 +344,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() - ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Builds: known, Logf: logger.Printf, + ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }} // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index d272cd7..285279f 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index e8660dd..b3dab4f 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index cd48a70..4189fc9 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -9,6 +9,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -47,7 +48,11 @@ type Indexer struct { Modules Store // Manifests is the manifests bucket; nil writes no manifests (issue 1449). Manifests Store - Builds *Builds + // Taken is the taken bucket: per repository the last merge taken in (taken.go, ADR 0556); nil writes none. + Taken Store + // Now is the clock of a taken value; nil is time.Now. + Now func() time.Time + Builds *Builds // Named asks the forge for the git object of each named path of the module directories given, at a commit // (gitea.gitea_commit_trees with paths; novox/hq ADR 0499 clauses 2 and 3): the indexer is the one reader of a // manifest's paths (ADR 0495 clause 1, the lead's decision L8 of R4e). Nil asks nothing, and a module naming paths @@ -98,6 +103,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { if err := ix.applyMerged(ctx, f); err != nil { return err } + if err := ix.writeTaken(ctx, f); err != nil { + return err + } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } if err := ix.applyPlacement(ctx, f); err != nil { diff --git a/modules/mesh-delivery-indexer/internal/index/index_test.go b/modules/mesh-delivery-indexer/internal/index/index_test.go index 74bcc57..3dd794d 100644 --- a/modules/mesh-delivery-indexer/internal/index/index_test.go +++ b/modules/mesh-delivery-indexer/internal/index/index_test.go @@ -17,8 +17,9 @@ const ( // memStore is the candidates bucket's compare-and-set behaviour in memory: a revision per write, a conflict when the // revision read is no longer the key's. Writes counts the writes that went through; beforeWrite lets a test change the -// key under the indexer. +// key under the indexer; unbound, when set, is every read's error (a state not yet on the bus). type memStore struct { + unbound error vals map[string][]byte revs map[string]uint64 next uint64 @@ -29,6 +30,9 @@ type memStore struct { func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} } func (m *memStore) Get(_ context.Context, key string) ([]byte, uint64, bool, error) { + if m.unbound != nil { + return nil, 0, false, m.unbound + } v, ok := m.vals[key] return v, m.revs[key], ok, nil } diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index 18987ef..71bf565 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -8,6 +8,7 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) const merge = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" @@ -94,22 +95,46 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value // writes is the values written by then; commit is each module's commit after. writes int commit map[string]string + // taken is the merge the repository's taken key names after then (issue 1280, ADR 0556 clause 1). + taken string }{ - {name: "the same merge again", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, + {name: "the same merge again", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { return ix.Apply(context.Background(), mergedMsg(repo, three)) }}, - {name: "a merge touching one of three modules", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, + {name: "a merge touching one of three modules", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next, then: func(ix *Indexer) error { m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"}) m.Merge = next return ix.Apply(context.Background(), m) }}, - {name: "a later source revision of another repository's merge", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, + {name: "a merge touching none of them is still taken in", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next, + then: func(ix *Indexer) error { + m := mergedMsg(repo, three) + m.Merge = next + return ix.Apply(context.Background(), m) + }}, + {name: "a merge while the taken state is not on the bus yet is still taken in", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: merge, + then: func(ix *Indexer) error { // issue 1790: the state is bound when it appears, so the indexer runs before it does + ix.Taken = &memStore{unbound: process.ErrNotYet} + m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"}) + m.Merge = next + return ix.Apply(context.Background(), m) + }}, + {name: "an earlier merge heard again after a later one leaves taken", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, + then: func(ix *Indexer) error { + ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == next && in == merge, nil } + m := mergedMsg(repo, three) + m.Merge = next + return ix.Apply(context.Background(), m) + }}, + {name: "a later source revision of another repository's merge", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge, then: func(ix *Indexer) error { return ix.ApplySource(context.Background(), 9, delivery.Source{Group: "g", Heads: map[string]string{repo: merge, "novox/mesh-catalog": otherMerge}}) }}, } { t.Run(tc.name, func(t *testing.T) { ix, mods := modulesIndexer(t) + taken := newMem() + ix.Taken = taken ctx := context.Background() if err := ix.Apply(ctx, mergedMsg(repo, three)); err != nil { t.Fatal(err) @@ -129,6 +154,10 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value t.Errorf("%s = %+v (%v), want commit %.8s", module, m, ok, want) } } + key, _ := delivery.RepositoryKey(repo) + if got, err := delivery.DecodeTaken(taken.vals[key]); err != nil || got.Merge != tc.taken || got.Repository != repo { + t.Errorf("taken.%s = %+v (%v), want merge %.8s", key, got, err, tc.taken) + } }) } } diff --git a/modules/mesh-delivery-indexer/internal/index/taken.go b/modules/mesh-delivery-indexer/internal/index/taken.go new file mode 100644 index 0000000..576b2a4 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/taken.go @@ -0,0 +1,60 @@ +package index + +import ( + "context" + "errors" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// What the indexer took in (novox/hq ADR 0556 clause 1, issue 1280): after a merged message is applied whole, the key of +// its repository in the state `taken` names that merge, so a reader of the modules knows they say the repository at that +// merge or later. The indexer is the key's only writer. A merge heard again after a later one (a redelivery) does not put +// the key back: the key moves only to a merge that contains the one it names, as the forge says it; a forge that cannot +// say keeps the key as it is and says so. + +// writeTaken writes the repository's taken key for the merge f carries, unless it names that merge or a later one. +func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error { + if ix.Taken == nil || f.Merge == "" { + return nil + } + key, err := delivery.RepositoryKey(f.Repository) + if err != nil { + ix.logf("%s at %s: no taken key is written: %v", f.Repository, f.Merge, err) + return nil + } + err = write(ctx, ix.Taken, key, func(value []byte, found bool) ([]byte, bool, error) { + if found { + if cur, err := delivery.DecodeTaken(value); err == nil { + if cur.Merge == f.Merge { + return nil, false, nil + } + if ix.Contains != nil { + later, err := ix.Contains(ctx, f.Repository, f.Merge, cur.Merge) + if err != nil { + ix.logf("%s: whether %s is in %s cannot be read, so taken.%s stays at %s: %v", f.Repository, f.Merge, cur.Merge, key, cur.Merge, err) + return nil, false, nil + } + if later { + return nil, false, nil // a merge heard again after a later one + } + } + } + } + now := time.Now + if ix.Now != nil { + now = ix.Now + } + data, err := delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName(f.Repository), Merge: f.Merge, At: now().UTC()}) + return data, false, err + }) + if errors.Is(err, process.ErrNotYet) { + // The state is not on the bus yet (a delivery that adds it, bound when it appears as issue 1790 says): said, and + // the merge's modules are still taken in; the next merge of the repository writes its key once the state appears. + ix.logf("%s at %s: its taken key is not written: %v", f.Repository, f.Merge, err) + return nil + } + return err +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 8becfb2..0122ca7 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -85,7 +85,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []string{"placement-changed"}; !reflect.DeepEqual(m.Emits, want) { t.Errorf("emits = %v, want %v", m.Emits, want) } - if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index ce8432b..04bca8c 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -19,6 +19,7 @@ { "name": "manifests", "history": 1 }, { "name": "modules", "history": 20 }, { "name": "placement", "history": 10 }, + { "name": "taken", "history": 1 }, { "name": "takeins", "history": 1 } ], "reads": [ diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index 40f1f20..a900626 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -286,16 +286,17 @@ func (t *Teller) catchUp(id string) { t.said(h, *w, err) } -// mergeAct is what the approve option of the gate's ask binds: 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 +// mergeAct is what the approve option of the gate's ask binds: the ask's verb (delivery.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. +// 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(ask, option string, heads map[string]string) asks.Act { +func mergeAct(verb, ask, option string, heads map[string]string) asks.Act { var out []string for id, sha := range heads { out = append(out, id+"="+sha) } sort.Strings(out) - return asks.Act{"verb": "mesh-delivery.merge", "ask": ask, "option": option, "heads": strings.Join(out, ",")} + return asks.Act{"verb": verb, "ask": ask, "option": option, "heads": strings.Join(out, ",")} } // headsOf is the heads an ask covers as an act names them: a delivery's id, `owner/repository@ 0 { + sum := sha256.Sum256([]byte(strings.Join(e.NotYet, ","))) + id += "." + hex.EncodeToString(sum[:6]) + } + } + if err := emitEvent(ctx, js, cred, DeliveredEventType, e, id); err != nil { logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err) } }) + model.SetReachWriter(func(w view.ReachWrite) { + if err := putReach(ctx, js, w); err != nil { + logger.Printf("the reach of %s could not be written (written again at its next change): %v", w.Key, err) + } + }) model.SetBacklog(view.BacklogOver(js, 5*time.Second)) - for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} { - go follow(ctx, js, bucket, model, logger) + for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} { + go follow(ctx, js, bucket, false, model, logger) + } + // The modules' and the source's history give each fingerprint's first commit and each head's source revision (ADR 0556). + for _, bucket := range []string{delivery.ModulesBucket, delivery.SourceBucket} { + go follow(ctx, js, bucket, true, model, logger) } select { case <-ctx.Done(): @@ -181,11 +199,21 @@ func (w kvWriter) Put(key string, value []byte) error { return err } +// putReach writes a reach value under its key in the view's own state reach (ADR 0556 clause 2), its only writer. +func putReach(ctx context.Context, js jetstream.JetStream, w view.ReachWrite) error { + kv, err := js.KeyValue(ctx, delivery.ReachBucket) + if err != nil { + return fmt.Errorf("the bucket %s: %w", delivery.ReachBucket, err) + } + _, err = kv.Put(ctx, w.Key, w.Value) + return err +} + // follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is // not there yet, or a watch that ends, is tried again. -func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model, logger *log.Logger) { +func follow(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model, logger *log.Logger) { for ctx.Err() == nil { - if err := watchOnce(ctx, js, bucket, model); err != nil && ctx.Err() == nil { + if err := watchOnce(ctx, js, bucket, history, model); err != nil && ctx.Err() == nil { logger.Printf("cannot watch %s, trying again: %v", bucket, err) } select { @@ -195,12 +223,17 @@ func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *v } } -func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model) error { +func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model) error { kv, err := js.KeyValue(ctx, bucket) if err != nil { return err } - w, err := kv.WatchAll(ctx) + var opts []jetstream.WatchOpt + initial := bucket + if history { + opts, initial = append(opts, jetstream.IncludeHistory()), view.ReachHistory(bucket) + } + w, err := kv.WatchAll(ctx, opts...) if err != nil { return err } @@ -214,11 +247,11 @@ func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model return errors.New("the watch ended") } if e == nil { // the last values have all been delivered - model.Initial(bucket) + model.Initial(initial) continue } model.Apply(view.Entry{Bucket: bucket, Key: e.Key(), Value: e.Value(), At: e.Created(), - Deleted: e.Operation() != jetstream.KeyValuePut, Revision: e.Revision()}) + Deleted: e.Operation() != jetstream.KeyValuePut, Revision: e.Revision(), History: history}) } } } diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 788bbc7..a1dec0c 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index e8660dd..b3dab4f 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-view/internal/view/reach.go b/modules/mesh-delivery-view/internal/view/reach.go new file mode 100644 index 0000000..d055f0d --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/reach.go @@ -0,0 +1,347 @@ +package view + +import ( + "fmt" + "maps" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// How far each repository reached (novox/hq ADR 0556 clauses 2 and 3, issue 1280). The view keeps, per repository the gate's +// source names, how far each of its modules reached on the nodes that want it, and derives the repository's reached head: the +// newest head of the repository in the source's history, not past the merge the indexer took in, at which every wanted module's +// floor is at or after its last change at or before that head. It writes that as its state reach and says each new reached +// head as its event delivered, per repository. Nothing here records a merge as delivered (ADR 0429 §1): a merge at or before +// the reached head is on the machines, derived by whoever reads it. + +// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' and the source's history +// give each fingerprint's first commit and each head's source revision. +func ReachHistory(bucket string) string { return bucket + "#history" } + +// reachBuckets are the watches the derivation waits for before it writes or says anything, so a start does not write a reach +// derived from half its inputs. +var reachBuckets = []string{delivery.ModulesBucket, delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.SourceBucket, + delivery.TakenBucket, ReachHistory(delivery.ModulesBucket), ReachHistory(delivery.SourceBucket)} + +// ReachWrite is a reach value to write under its key in the view's state reach. +type ReachWrite struct { + Key string + Value []byte +} + +// SetReachWriter sets where the view's reach values go; nothing is written until it is set. +func (m *Model) SetReachWriter(put func(ReachWrite)) { + m.mu.Lock() + defer m.mu.Unlock() + m.reachPut = put +} + +// applyHistory takes one revision of a watch with history: of the indexer's modules (each module's values in order) and of +// the gate's source (each revision's heads). +func (m *Model) applyHistory(e Entry) { + switch e.Bucket { + case delivery.ModulesBucket: + if e.Deleted { + delete(m.modHist, e.Key) + return + } + mod, err := delivery.DecodeModule(e.Value) + if err != nil || mod.FP2 == "" || mod.Commit == "" { + return + } + h := m.modHist[e.Key] + if n := len(h); n > 0 && h[n-1].FP2 == mod.FP2 && h[n-1].Commit == mod.Commit { + if mod.Src != 0 { + h[n-1].Src = mod.Src + } + return + } + h = append(h, mod) + if len(h) > 2*delivery.MostChanges { + h = h[len(h)-delivery.MostChanges:] + } + m.modHist[e.Key] = h + case delivery.SourceBucket: + if e.Key != delivery.SourceMainKey || e.Deleted || e.Revision == 0 { + return + } + src, err := delivery.DecodeSource(e.Value) + if err != nil { + return + } + m.srcHist[e.Revision] = src.Heads + } +} + +// headAt is one head of a repository and the first source revision that named it. +type headAt struct { + commit string + rev uint64 +} + +// headsOf are the heads of a repository in the source's history, oldest first, each with the first revision naming it. +func (m *Model) headsOf(repository string) []headAt { + var out []headAt + for _, rev := range slices.Sorted(maps.Keys(m.srcHist)) { + for r, head := range m.srcHist[rev] { + if !delivery.SameRepository(r, repository) || head == "" { + continue + } + if n := len(out); n == 0 || !sameCommit(out[n-1].commit, head) { + out = append(out, headAt{commit: head, rev: rev}) + } + } + } + return out +} + +// revOf is the first source revision naming a commit as the repository's head, or 0 when the history holds none. +func revOf(heads []headAt, commit string) uint64 { + for _, h := range heads { + if sameCommit(h.commit, commit) { + return h.rev + } + } + return 0 +} + +// sameCommit is whether two spellings of a commit are one (either may be abbreviated to at least seven characters). +func sameCommit(a, b string) bool { + a, b = strings.ToLower(a), strings.ToLower(b) + if len(a) > len(b) { + a, b = b, a + } + return len(a) >= 7 && strings.HasPrefix(b, a) +} + +// noBound is a module's bound when its floor is its newest change: it holds no head back. +const noBound = ^uint64(0) + +// wantKeys are the wants' keys by module, each sorted: read once for every repository a derivation looks at. +func (m *Model) wantKeys() map[string][]string { + out := map[string][]string{} + for _, key := range slices.Sorted(maps.Keys(m.wants)) { + _, module := splitKey(key) + out[module] = append(out[module], key) + } + return out +} + +// deriveReach is the reach of one repository from the model as it stands; wants are the wants' keys by module (wantKeys). +func (m *Model) deriveReach(repository string, wants map[string][]string) delivery.Reach { + r := delivery.Reach{Repository: delivery.RepositoryName(repository), Modules: map[string]delivery.ModuleReach{}} + heads := m.headsOf(repository) + key, _ := delivery.RepositoryKey(repository) + if t, ok := m.taken[key]; ok { + r.Indexed = t.Merge + } + bound := noBound + var holds []string + for _, name := range slices.Sorted(maps.Keys(m.modules)) { + me := m.modules[name] + if me.err != "" || me.value.FP2 == "" || !delivery.SameRepository(me.value.Repository, repository) { + continue + } + mr, b, words := m.moduleReach(name, me.value, heads, wants[name]) + r.Modules[name] = mr + if b < bound { + bound = b + } + if b != noBound { + holds = append(holds, words) + } + } + if len(r.Modules) == 0 { + r.Modules = nil + } + indexed := uint64(0) + if r.Indexed != "" { + indexed = revOf(heads, r.Indexed) + } + switch { + case r.Indexed == "": + holds = append(holds, "the indexer has taken in no merge of "+r.Repository) + case indexed == 0: + holds = append(holds, fmt.Sprintf("the merge the indexer took in, %.12s, is not a head of %s in the source's history", r.Indexed, r.Repository)) + } + for _, h := range heads { + if h.rev <= indexed && h.rev < bound { + r.Reached, r.ReachedSrc = h.commit, h.rev + } + } + if n := len(heads); n > 0 && !sameCommit(heads[n-1].commit, r.Reached) { + if indexed != 0 && indexed < heads[n-1].rev { + holds = append(holds, fmt.Sprintf("the indexer has not yet taken in %.12s", heads[n-1].commit)) + } + slices.Sort(holds) + r.Holds = slices.Compact(holds) + } + return r +} + +// moduleReach is how far one module reached, the source revision it holds the reached head before (noBound when none), and +// what holds it, in words. A node counts for the fingerprint its running have runs, healthy or not (ADR 0556 clause 2). +func (m *Model) moduleReach(name string, mod delivery.Module, heads []headAt, wantKeys []string) (delivery.ModuleReach, uint64, string) { + mr := delivery.ModuleReach{FP: mod.FP2} + hist := m.modHist[name] + if n := len(hist); n == 0 || hist[n-1].FP2 != mod.FP2 || hist[n-1].Commit != mod.Commit { + hist = append(slices.Clone(hist), mod) // the value watched now is the newest, whatever the history watch has given + } + for i := len(hist) - 1; i >= 0 && len(mr.Changes) < delivery.MostChanges; i-- { + c := delivery.Change{Commit: hist[i].Commit, FP: hist[i].FP2, Src: hist[i].Src} + if c.Src == 0 { + c.Src = revOf(heads, c.Commit) + } + if n := len(mr.Changes); n > 0 && mr.Changes[n-1].FP == c.FP { + mr.Changes[n-1] = c // the same fingerprint again, back to back: its first appearance is the older commit + continue + } + mr.Changes = append(mr.Changes, c) + } + var waiting []string + floor, floorNode, unset := -1, "", "" + for _, key := range wantKeys { + node, _ := splitKey(key) + w := m.wants[key] + if w.err != "" || w.value.Removed || w.value.FP == "" { + continue + } + mr.Nodes++ + h, ok := m.haves[key] + switch { + case !ok || h.err != "" || h.have.FP == "": + waiting = append(waiting, fmt.Sprintf("%s on %s runs nothing yet", name, node)) + unset = waiting[len(waiting)-1] + continue + case !runs(h.have.Phase): + waiting = append(waiting, fmt.Sprintf("%s on %s is %s at %s, not running", name, node, h.have.Phase, short(h.have.FP))) + unset = waiting[len(waiting)-1] + continue + case h.have.FP != w.value.FP: + waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet the %s it wants", name, node, short(h.have.FP), short(w.value.FP))) + case h.have.Health == delivery.HealthFailed: + waiting = append(waiting, fmt.Sprintf("%s on %s runs %s and its health check failed", name, node, short(h.have.FP))) + case h.have.Health != delivery.HealthOK: + waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet healthy", name, node, short(h.have.FP))) + } + i := slices.IndexFunc(mr.Changes, func(c delivery.Change) bool { return c.FP == h.have.FP }) + if i < 0 { + unset = fmt.Sprintf("%s on %s runs %s, which is not among its %d known changes", name, node, short(h.have.FP), len(mr.Changes)) + waiting = append(waiting, unset) + continue + } + if i > floor { + floor, floorNode = i, node + } + } + mr.Waiting = strings.Join(waiting, "; ") + switch { + case mr.Nodes == 0: + return mr, noBound, "" + case unset != "": + // A node that wants the module runs nothing countable of it (no have, not running, or a build older than the kept + // history): the module holds no head of the repository back, which would make every apply anywhere, or one module + // stuck failed, freeze the whole repository (review of issue 1280, ADR 0556 clause 2). Its words name it, and a merge + // among its changes is not yet on the machines (notYet). + return mr, noBound, "" + } + mr.Floor = mr.Changes[floor].Commit + if floor == 0 { + return mr, noBound, "" + } + next := mr.Changes[floor-1] + words := fmt.Sprintf("%s on %s runs the build of %.12s, not yet that of %.12s", name, floorNode, mr.Floor, next.Commit) + if next.Src == 0 { + // The next change's order is not known: the head is held at the floor's own revision, or before every head when that is + // not known either. + if own := mr.Changes[floor].Src; own != 0 { + return mr, own + 1, words + } + return mr, 0, words + } + return mr, next.Src, words +} + +// runs is whether a have in this phase runs the fingerprint it names: running, being checked after its apply, or held after a +// failed backup, which keeps running what it has (ADR 0539 clause 16). A have backing up or applying, failed or removed runs +// nothing the derivation can count. +func runs(p delivery.Phase) bool { + return p == delivery.PhaseRunning || p == delivery.PhaseChecking || p == delivery.PhaseHeld +} + +// short is a fingerprint's first twelve characters, after its form. +func short(fp string) string { + fp = strings.TrimPrefix(fp, delivery.SourcePrefix2) + if len(fp) > 12 { + fp = fp[:12] + } + return fp +} + +// reachReady is whether every watch the derivation reads handed over what stood at its start. +func (m *Model) reachReady() bool { + for _, b := range reachBuckets { + if !m.initial[b] { + return false + } + } + return m.source != nil && m.source.err == "" +} + +// reachNow derives the reach of every repository the source names, and answers the values that changed and the reached +// heads not yet said. It is called with the lock held. +func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) { + if (m.reachPut == nil && m.deliveredEvent == nil) || !m.reachReady() { + return nil, nil + } + var writes []ReachWrite + var events []DeliveredEvent + wants := m.wantKeys() + for _, repository := range slices.Sorted(maps.Keys(m.source.heads)) { + key, err := delivery.RepositoryKey(repository) + if err != nil { + continue + } + r := m.deriveReach(repository, wants) + if prev, ok := m.reach[key]; !ok || !delivery.SameReach(prev, r) { + if data, err := delivery.EncodeReach(r); err == nil { + m.reach[key] = r + writes = append(writes, ReachWrite{Key: key, Value: data}) + } + } + notYet := notYetOf(r) + said := strings.Join(notYet, ",") + if r.Reached != "" && (r.ReachedSrc > m.reachSaid[key] || said != m.reachNotYet[key]) { + m.reachSaid[key], m.reachNotYet[key] = r.ReachedSrc, said + events = append(events, DeliveredEvent{Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached}, + Repository: r.Repository, NotYet: notYet}) + } + } + return writes, events +} + +// notYetOf are the merges at or before a repository's reached head that are not yet on the machines: the changes of each module +// a node wants with no floor (a node runs nothing countable of it), sorted. A consumer takes every other merge at or before +// the reached head as on the machines. +func notYetOf(r delivery.Reach) []string { + var out []string + for _, mr := range r.Modules { + if mr.Nodes == 0 || mr.Floor != "" { + continue + } + for _, c := range mr.Changes { + out = append(out, c.Commit) + } + } + slices.Sort(out) + return slices.Compact(out) +} + +// Reach is the reach of every repository as the view derived it last, by key. +func (m *Model) Reach() map[string]delivery.Reach { + m.mu.Lock() + defer m.mu.Unlock() + return maps.Clone(m.reach) +} diff --git a/modules/mesh-delivery-view/internal/view/reach_test.go b/modules/mesh-delivery-view/internal/view/reach_test.go new file mode 100644 index 0000000..f459efe --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/reach_test.go @@ -0,0 +1,157 @@ +package view + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1280 (ADR 0556 clauses 2 and 3): the view derives, per repository, how far each module reached and the reached head, writes +// it when it differs, and says each new reached head once as its event delivered. +func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { + const web, sdk = "novox/web", "novox/sdk" + commitA, commitB, commitC := strings.Repeat("a", 40), strings.Repeat("b", 40), strings.Repeat("c", 40) + commitS, commitX := strings.Repeat("5", 40), strings.Repeat("e", 40) + const fpMail = "src2:3333333333333333333333333333333333333333333333333333333333333333" + mod := func(module, fp, commit string, src uint64, history bool) Entry { + v, err := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: strings.Repeat("a", 64), Repository: web, Commit: commit, Src: src, Path: module}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.ModulesBucket, Key: module, Value: v, At: t0, History: history} + } + src := func(rev uint64, webHead string) Entry { + v, err := delivery.EncodeSource(delivery.Source{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.SourceBucket, Key: delivery.SourceMainKey, Value: v, At: t0, Revision: rev, History: true} + } + taken := func(repository, merge string) Entry { + key, _ := delivery.RepositoryKey(repository) + v, err := delivery.EncodeTaken(delivery.Taken{Repository: repository, Merge: merge, At: t0}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.TakenBucket, Key: key, Value: v, At: t0} + } + haveIn := func(node, module, fp string, phase delivery.Phase, health delivery.Health) Entry { + e := haveFP(t, node, module, fp, health) + v, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: phase, Health: health}) + e.Value = v + return e + } + // 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, _ := delivery.DecodeSource(src(12, commitC).Value) + current := Entry{Bucket: delivery.SourceBucket, Key: delivery.SourceMainKey, Revision: 12, At: t0} + current.Value, _ = delivery.EncodeSource(cur) + return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current, + mod("web", fpOld, commitA, 10, true), mod("web", fpNew, commitC, 12, true), mod("mail", fpMail, commitA, 0, true), + mod("web", fpNew, commitC, 12, false), mod("mail", fpMail, commitA, 0, false), + taken(web, commitC), taken(sdk, commitS), + wantE(t, "ace", "web", delivery.Want{FP: fpNew}), haveFP(t, "ace", "web", fpNew, delivery.HealthOK), + wantE(t, "g14", "web", delivery.Want{FP: fpNew}), haveFP(t, "g14", "web", fpNew, delivery.HealthOK), + wantE(t, "ace", "mail", delivery.Want{FP: fpMail}), haveFP(t, "ace", "mail", fpMail, delivery.HealthOK)} + } + for _, tc := range []struct { + name string + then []Entry + reached string // web's reached head; "" none + holds string // words one of web's holds carries; "" none + waiting string // words web's module web waits with; "" none + notYet []string // the merges the reached head's event says are not yet on the machines + }{ + {name: "every node runs the newest build", reached: commitC}, + {name: "a node lagging stops the head before the change it lacks", reached: commitB, holds: "web on g14", + then: []Entry{haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}, waiting: "not yet the"}, + {name: "a want held back at an older build holds the head as a lag does", reached: commitB, holds: "web on g14 runs the build of", + then: []Entry{wantE(t, "g14", "web", delivery.Want{FP: fpOld, Newest: fpNew, Rule: 2}), haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}}, + {name: "a failing health check is said and holds nothing", reached: commitC, waiting: "health check failed", + then: []Entry{haveFP(t, "g14", "web", fpNew, delivery.HealthFailed)}}, + {name: "a node with no running have holds no head back and its module's merges are not yet", reached: commitC, waiting: "not running", + notYet: []string{commitA, commitC}, + then: []Entry{haveIn("g14", "web", fpNew, delivery.PhaseApplying, delivery.HealthUnknown)}}, + {name: "an unrelated module not running leaves the head reached (the reviewer's probe)", reached: commitC, + notYet: []string{commitA}, + then: []Entry{haveIn("ace", "mail", fpMail, delivery.PhaseFailed, delivery.HealthFailed)}}, + {name: "a module no node wants holds nothing", reached: commitC, + then: []Entry{wantE(t, "ace", "mail", delivery.Want{Removed: true})}}, + {name: "the indexer not yet having taken the head in", reached: commitB, holds: "has not yet taken in", + then: []Entry{taken(web, commitB)}}, + {name: "a change whose source revision is not known yet holds no head the source named", reached: commitC, + then: []Entry{mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, true), mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, false)}}, + } { + t.Run(tc.name, func(t *testing.T) { + m := New() + var written []ReachWrite + var said []DeliveredEvent + m.SetReachWriter(func(w ReachWrite) { written = append(written, w) }) + m.SetDeliveredEvent(func(e DeliveredEvent) { + if e.Repository != "" { + said = append(said, e) + } + }) + fill(m, world(t)...) + if len(written) != 0 { + t.Fatalf("written before every watch handed over: %v", written) + } + for _, b := range reachBuckets { + m.Initial(b) + } + fill(m, tc.then...) + key, _ := delivery.RepositoryKey(web) + r := m.Reach()[key] + if r.Reached != tc.reached { + t.Errorf("reached = %.8s, want %.8s (holds %v)", r.Reached, tc.reached, r.Holds) + } + if got := strings.Join(r.Holds, "; "); (tc.holds == "") != (got == "") || !strings.Contains(got, tc.holds) { + t.Errorf("holds = %q, want words with %q", got, tc.holds) + } + if got := r.Modules["web"].Waiting; (tc.waiting == "") != (got == "") || !strings.Contains(got, tc.waiting) { + t.Errorf("web waits with %q, want words with %q", got, tc.waiting) + } + // A library reaches the head the indexer took in, with no module. + sk, _ := delivery.RepositoryKey(sdk) + if lib := m.Reach()[sk]; lib.Reached != commitS || len(lib.Modules) != 0 { + t.Errorf("the library's reach = %+v", lib) + } + // Each key is written once with its last value; a reached head is said once, for its repository alone. + last := map[string]delivery.Reach{} + for _, w := range written { + v, err := delivery.DecodeReach(w.Value) + if err != nil { + t.Fatalf("written %s: %v", w.Key, err) + } + last[w.Key] = v + } + if !delivery.SameReach(last[key], r) { + t.Errorf("the last value written is %+v, not the reach derived %+v", last[key], r) + } + if tc.reached != "" { + var last DeliveredEvent + for _, e := range said { + if e.Repository == web { + last = e + } + } + if strings.Join(last.NotYet, ",") != strings.Join(tc.notYet, ",") { + t.Errorf("not yet = %v, want %v", last.NotYet, tc.notYet) + } + } + { + n := len(said) + fill(m, tc.then...) + if len(said) != n && tc.reached == commitC { + t.Errorf("said again: %+v", said[n:]) + } + for _, e := range said { + if len(e.Heads) != 1 || e.Group != "" || e.Revision == 0 { + t.Errorf("a reached head said as %+v", e) + } + } + } + }) + } +} diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 4ca99eb..0aea653 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -41,9 +41,41 @@ type Reader struct { WantsState StateRead // the assigner's wants and holds DifferencesState StateRead // the assigner's differences HavesState StateRead // the node-engine's haves + ReachState StateRead // the view's own reach (ADR 0556) PlacementState StateRead // the indexer's placement } +// ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556). +const ReachReadTool = "view_reach" + +// Reach is the stored reach of one repository, or of every repository when none is named, as the view wrote it. +func (r Reader) Reach(repository string) (any, error) { + if r.ReachState.Get == nil || r.ReachState.Keys == nil { + return nil, fmt.Errorf("the reach state cannot be read here") + } + if repository = strings.TrimSpace(repository); repository != "" { + key, err := delivery.RepositoryKey(repository) + if err != nil { + return nil, err + } + raw, err := r.ReachState.Get(key) + if err != nil { + return nil, err + } + if len(raw) == 0 { + return nil, fmt.Errorf("no reach of %s: the view has derived none (the gate's source does not name it, or the view waits for a watch)", repository) + } + return raw, nil + } + keys, err := r.ReachState.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + values, err := r.ReachState.values(keys) + return map[string]any{"reach": values}, err +} + // Module is the stored value of modules., exactly as the indexer wrote it. func (r Reader) Module(module string) (json.RawMessage, error) { module = strings.TrimSpace(module) diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 8e3ecff..2c68ba2 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -23,6 +23,8 @@ type Entry struct { Deleted bool // Revision is the bucket revision of the change, when the watch gives one (the source revision is the gate's source key's). Revision uint64 + // History says the change comes from a watch with history (ReachHistory): it feeds only the history the reach reads. + History bool } // Row is one module on one node. @@ -240,6 +242,14 @@ type Model struct { // the watchman's buckets, by key presence map[string]presenceEntry stuck map[string]stuckEntry + // how far each repository reached (reach.go, ADR 0556) + modHist map[string][]delivery.Module // the indexer's values of each module, oldest first + srcHist map[uint64]map[string]string // the source's heads, by revision + taken map[string]delivery.Taken // the indexer's taken, by repository key + reach map[string]delivery.Reach // the reach derived last, by repository key + reachSaid map[string]uint64 // the reached head's source revision said last, by repository key + reachNotYet map[string]string // the merges said not yet on the machines with it, by repository key + reachPut func(ReachWrite) } type presenceEntry struct { @@ -348,21 +358,41 @@ type nodeEntry struct { func New() *Model { return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]delivery.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{}, haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}, - presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}} + presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}, + modHist: map[string][]delivery.Module{}, srcHist: map[uint64]map[string]string{}, taken: map[string]delivery.Taken{}, + reach: map[string]delivery.Reach{}, reachSaid: map[string]uint64{}, reachNotYet: map[string]string{}} } // Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes // the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released. func (m *Model) Apply(e Entry) { m.mu.Lock() - m.apply(e) - event := m.deliveredNow() - say := m.deliveredEvent - motion, sayMotion := m.motionNow(e), m.motionEvent + var event *DeliveredEvent + var motion []MotionEvent + if e.History { + m.applyHistory(e) + } else { + m.apply(e) + event = m.deliveredNow() + motion = m.motionNow(e) + } + writes, reached := m.reachNow() + say, put := m.deliveredEvent, m.reachPut + sayMotion := m.motionEvent m.mu.Unlock() if event != nil && say != nil { say(*event) } + if put != nil { + for _, w := range writes { + put(w) + } + } + if say != nil { + for _, ev := range reached { + say(ev) + } + } if sayMotion != nil { for _, ev := range motion { sayMotion(ev) @@ -373,10 +403,18 @@ func (m *Model) Apply(e Entry) { // DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR // 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A // consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view). +// +// The same event says a repository's reached head (ADR 0556 clause 3): Repository names it, Heads holds its head alone, Revision +// is the source revision that named that head and Group is empty; a consumer takes every earlier merge of that repository as on +// the machines. type DeliveredEvent struct { - Revision uint64 `json:"revision"` - Group string `json:"group"` - Heads map[string]string `json:"heads"` + Revision uint64 `json:"revision"` + Group string `json:"group"` + Heads map[string]string `json:"heads"` + Repository string `json:"repository,omitempty"` + // NotYet are, in the per-repository form, the merges at or before the reached head that are not yet on the machines: the + // changes of a module a node wants and runs nothing countable of. Said again when they change. + NotYet []string `json:"not_yet,omitempty"` } // SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set. @@ -512,6 +550,16 @@ func (m *Model) apply(e Entry) { } d, err := delivery.DecodeComposedDigest(e.Value) m.composed[e.Key] = composedEntry{value: d, err: errText(err)} + case delivery.TakenBucket: + if e.Deleted { + delete(m.taken, e.Key) + return + } + if t, err := delivery.DecodeTaken(e.Value); err == nil { + m.taken[e.Key] = t + } else { + delete(m.taken, e.Key) + } case delivery.SourceBucket: if e.Key != delivery.SourceMainKey { return @@ -1214,10 +1262,22 @@ func (m *Model) Initial(bucket string) { } } say := m.motionEvent + writes, reached := m.reachNow() + put, sayDelivered := m.reachPut, m.deliveredEvent m.mu.Unlock() for _, ev := range out { say(ev) } + if put != nil { + for _, w := range writes { + put(w) + } + } + if sayDelivered != nil { + for _, ev := range reached { + sayDelivered(ev) + } + } } // motionOf is the events to say now about the node and module of a model key, from their want, have and stuck key. diff --git a/modules/mesh-delivery-view/internal/view/view_test.go b/modules/mesh-delivery-view/internal/view/view_test.go index d6ce64e..d228cab 100644 --- a/modules/mesh-delivery-view/internal/view/view_test.go +++ b/modules/mesh-delivery-view/internal/view/view_test.go @@ -110,7 +110,7 @@ func TestWaitsForTheNodeAndForANewerBuild(t *testing.T) { // 1134/AC2 } } -func TestLeftBehindIsNotRunning(t *testing.T) { // 1134/AC3 +func TestLeftBehindIsNotRunningAndPulledInIs(t *testing.T) { // 1134/AC3 (both halves) for name, extra := range map[string]string{"ended": `,"ended":"2026-10-18T11:30:00Z"`, "state": `,"state":"pending"`} { m := New() fill(m, base(t)...) @@ -131,13 +131,11 @@ func TestLeftBehindIsNotRunning(t *testing.T) { // 1134/AC3 if find(t, s, "a", "w").Status != StatusLeftBehind || s.Counted() != 4 { t.Fatalf("a have with no assignment: %+v counted %d", find(t, s, "a", "w"), s.Counted()) } -} - -func TestPulledInIsRunning(t *testing.T) { // 1134/AC3 - m := New() + // A module pulled in by another stands: shown and counted, with what required it. + m = New() fill(m, base(t)...) fill(m, assign("b", "z", "b3", `,"pulled-in":true,"required-by":"x"`)) - s := m.Snapshot() + s = m.Snapshot() r := find(t, s, "b", "z") if r.Status != StatusAssigned || r.RequiredBy != "x" || r.Have != "b3" || s.Counted() != 4 { t.Fatalf("pulled-in row = %+v counted %d, want shown and counted with required-by x", r, s.Counted()) diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index b887e2b..1758584 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -84,7 +84,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 + want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { @@ -103,8 +103,9 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if len(m.Uses) != 0 { t.Errorf("uses = %v, want none", m.Uses) } - // The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads. - if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) { + // The states the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the + // teller reads, and how far each repository reached (ADR 0556 clause 2, issue 1280), which the gate reads. + if want := []state{{Name: "answers", History: 5}, {Name: "reach", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -115,7 +116,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. - if want := []string{"indexer_module", "indexer_modules", "indexer_placement", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"indexer_module", "indexer_modules", "indexer_placement", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage", "view_reach"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } for _, a := range m.Build.Artifacts { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index a99a8f6..41f42f7 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -15,7 +15,8 @@ "assigner_want", "assigner_differences", "node_haves", - "build_usage" + "build_usage", + "view_reach" ], "emits": [ "delivered", @@ -24,7 +25,8 @@ "stuck" ], "state": [ - { "name": "answers", "history": 5 } + { "name": "answers", "history": 5 }, + { "name": "reach", "history": 1 } ], "reads": [ "mesh-controller.assignments", @@ -47,6 +49,7 @@ "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", + "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck",