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 bf432663..5d60a336 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -435,6 +435,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: awaitedStore{b: graph}, SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, Message: messageVia(calls.CallKept), Revisions: &busObjects{js: js, local: index.CatalogRevisionsObjects}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) + SayCatalogRevision: sayCatalogRevision(conn, cred), // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go index abc08a3c..bb6ae703 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/objects.go @@ -3,13 +3,19 @@ package main import ( "bytes" "context" + "encoding/json" "errors" "fmt" "sync" + "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-indexer/internal/index" ) @@ -34,8 +40,11 @@ func (b *busObjects) open(ctx context.Context) (jetstream.ObjectStore, error) { return b.o, nil } o, err := b.js.ObjectStore(ctx, catalogue.ObjectsBucket(indexerModule, b.local)) + if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) { + return nil, fmt.Errorf("the object bucket %s is not on the bus yet (the grants part creates it from the manifest's objects): %w", b.local, process.ErrNotYet) + } if err != nil { - return nil, fmt.Errorf("the object bucket %s is not open (the grants part creates it from the manifest's objects): %w", b.local, err) + return nil, fmt.Errorf("the object bucket %s is not open: %w", b.local, err) } b.o = o return o, nil @@ -107,3 +116,24 @@ func (b *busObjects) Fill(ctx context.Context) (int64, int64, error) { } return int64(st.Size()), info.CachedInfo().Config.MaxBytes, nil } + +// sayCatalogRevision publishes the seat mesh-catalog's event catalog-revision under the duplicate id of its kind and +// number, so a revision said twice is heard once (design 47, Two graphs). +func sayCatalogRevision(conn *nats.Conn, cred process.Account) func(context.Context, meshcatalog.CatalogRevisionEvent) error { + return func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error { + data, err := json.Marshal(e) + if err != nil { + return err + } + msg := nats.NewMsg(seatEventSubject(meshcatalog.EventCatalogRevision)) + id := fmt.Sprintf("%s-%s-%d%s", meshcatalog.EventCatalogRevision, e.Kind, e.Revision, e.Heads) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", cred.Module) + msg.Header.Set("x-node", cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Header.Set(nats.MsgIdHdr, id) + msg.Data = data + return conn.PublishMsg(msg) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 8e9c5b81..57d8f35d 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -100,6 +100,8 @@ type Indexer struct { // Revisions is the object bucket catalog-revisions (catalog_revisions.go; design 47, Two graphs); nil keeps no // catalog graph revision. Revisions Objects + // SayCatalogRevision says the seat mesh-catalog's event catalog-revision for a revision kept; nil says nothing. + SayCatalogRevision func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error // Graph is the module graph's state (GraphBucket, graph.go; novox/hq ADR 0603 §2); nil writes no graph. Graph Store // Files asks the forge for every blob under each directory at a commit, by directory then path relative to it, each diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go index 3a457422..ccb52f58 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go @@ -5,10 +5,13 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strconv" "strings" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/sdk/graph" + "git.novox.be/novox/mesh-sdk/go/process" ) // The indexer's object bucket catalog-revisions (design 47, Two graphs; novox/hq ADR 0619 §2: declared in the old @@ -118,3 +121,96 @@ func ReadRevision(ctx context.Context, o Objects, name string) (meshcatalog.Cata } return rev, entries, nil } + +// newestTrunk is the newest trunk revision the bucket keeps, its list only; none is a zero revision and false. +func newestTrunk(ctx context.Context, o Objects) (meshcatalog.CatalogRevision, bool, error) { + names, err := o.Names(ctx) + if err != nil { + return meshcatalog.CatalogRevision{}, false, err + } + var newest int64 + for _, n := range names { + if r, ok := TrunkRevisionOf(n); ok && r > newest { + newest = r + } + } + if newest == 0 { + return meshcatalog.CatalogRevision{}, false, nil + } + data, err := o.Get(ctx, "trunk."+strconv.FormatInt(newest, 10)) + if err != nil { + return meshcatalog.CatalogRevision{}, false, err + } + var rev meshcatalog.CatalogRevision + if err := json.Unmarshal(data, &rev); err != nil { + return rev, false, fmt.Errorf("trunk.%d: %w", newest, err) + } + return rev, true, nil +} + +// changedEntries are the modules whose entry differs between two revisions' lists, one gone or added included, by name. +func changedEntries(before, after []meshcatalog.EntryRef) []string { + was := map[string]string{} + for _, r := range before { + was[r.Module] = r.Digest + } + var changed []string + for _, r := range after { + if d, ok := was[r.Module]; !ok || d != r.Digest { + changed = append(changed, r.Module) + } + delete(was, r.Module) + } + for m := range was { + changed = append(changed, m) + } + slices.Sort(changed) + return changed +} + +// keepTrunkRevision keeps the catalog graph revision a merge's or a recompute's module graph makes (design 47, Two +// graphs: a revision for each source revision taken in, and for a fingerprint moved with no merge), numbered after the +// newest kept, and says catalog-revision; a graph whose entries are the newest revision's makes none. A revision not +// kept is said in the log and returned, so the take-in that made it is not said taken. +func (ix *Indexer) keepTrunkRevision(ctx context.Context, edges map[string]graphEdges, whole graph.Graph) error { + if ix.Revisions == nil { + return nil + } + entries := make(map[string]meshcatalog.CatalogEntry, len(edges)) + for name, e := range edges { + entries[name] = CatalogEntryOf(name, e.cur, whole[name]) + } + ix.mu.Lock() + src := int64(ix.source.rev) + ix.mu.Unlock() + newest, any, err := newestTrunk(ctx, ix.Revisions) + if errors.Is(err, process.ErrNotYet) { + ix.logf("catalog graph: no revision is kept: its objects %s are not on the bus yet", CatalogRevisionsObjects) + return nil + } + if err != nil { + ix.logf("catalog graph: no revision is kept: the newest is not read: %v", err) + return fmt.Errorf("the newest catalog graph revision: %w", err) + } + at := ix.now().UTC() + rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: newest.Revision + 1, Kind: "trunk", Source: src, At: &at}, entries) + if err != nil { + return err + } + changed := changedEntries(newest.Entries, rev.Entries) + if any && len(changed) == 0 { + return nil + } + if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil { + ix.logf("catalog graph: revision %d of source revision %d is not kept: %v", rev.Revision, src, err) + return err + } + ix.logf("catalog graph: revision %d of source revision %d kept, %d entries, %d changed", rev.Revision, src, len(rev.Entries), len(changed)) + if ix.SayCatalogRevision != nil { + if err := ix.SayCatalogRevision(ctx, meshcatalog.CatalogRevisionEvent{V: CatalogEntryLayout, Revision: rev.Revision, Kind: rev.Kind, + Source: src, At: &at, Changed: changed, Entries: int64(len(rev.Entries))}); err != nil { + ix.logf("catalog graph: revision %d is not said: %v", rev.Revision, err) + } + } + return nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/graph.go b/modules/mesh-delivery-indexer/internal/index/graph.go index 199f9f7d..2d37097f 100644 --- a/modules/mesh-delivery-indexer/internal/index/graph.go +++ b/modules/mesh-delivery-indexer/internal/index/graph.go @@ -81,22 +81,7 @@ func (ix *Indexer) writeGraph(ctx context.Context, edges map[string]graphEdges) GraphMost, len(names), names[GraphMost], len(names)-GraphMost-1) names = names[:GraphMost] } - providers := func(provision string) []catalogue.Manifest { - var out []catalogue.Manifest - for _, name := range slices.Sorted(maps.Keys(edges)) { - if slices.Contains(edges[name].manifest.Offers(), provision) { - out = append(out, edges[name].manifest) - } - } - return out - } - // The verb edges on a cycle are left out over the whole graph (ADR 0614 §3), every module it knows, so a module past - // GraphMost still closes the cycle it lies on. - whole := graph.Graph{} - for name, e := range edges { - whole[name] = graphNode(e, providers) - } - whole.LeaveOut() + whole := wholeGraph(edges) for _, name := range names { key, err := graph.Key(name) if err != nil { @@ -141,3 +126,23 @@ func (ix *Indexer) writeGraph(ctx context.Context, edges map[string]graphEdges) } return nil } + +// wholeGraph is the module graph of every module of edges, the verb edges on a cycle left out over the whole graph +// (ADR 0614 §3), so a module past GraphMost still closes the cycle it lies on. +func wholeGraph(edges map[string]graphEdges) graph.Graph { + providers := func(provision string) []catalogue.Manifest { + var out []catalogue.Manifest + for _, name := range slices.Sorted(maps.Keys(edges)) { + if slices.Contains(edges[name].manifest.Offers(), provision) { + out = append(out, edges[name].manifest) + } + } + return out + } + whole := graph.Graph{} + for name, e := range edges { + whole[name] = graphNode(e, providers) + } + whole.LeaveOut() + return whole +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 242886a9..4756fd1e 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -886,7 +886,7 @@ func (ix *Indexer) dry() *dryRun { Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil, Holds: nil, Counted: nil, AskEngines: nil, HoldBounds: nil, OwnBuildState: nil, underSeats: true, // A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep. - Revisions: nil} + Revisions: nil, SayCatalogRevision: nil} x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } x.Store, d.candidates = over(ix.Store) x.Modules, d.modules = over(ix.Modules) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index efa3a483..6a82b105 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -970,7 +970,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", - "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Holds", "Counted", "AskEngines", "HoldBounds", "OpenTries", "TryRecovery", "OwnBuildState", "Revisions", "SayCatalogRevision"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index 2ffb0617..0f8b0f29 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -372,7 +372,10 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc if err := ix.writeOverlay(ctx, overlay, finals, w); err != nil { return err } - return ix.writeGraph(ctx, edges) + if err := ix.writeGraph(ctx, edges); err != nil { + return err + } + return ix.keepTrunkRevision(ctx, edges, wholeGraph(edges)) } // writeOverlay writes the records of a merge, each once and only if it differs from what the bucket holds; finals has diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index 73f7a37d..5b9821dc 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -332,16 +332,34 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // if _, _, err := ReadRevision(context.Background(), kept, "trunk.2"); err == nil || !strings.Contains(err.Error(), next.Entries[0].Module) { t.Errorf("trunk.2 with a lost entry read without error naming it: %v", err) } + // W4.3: a recompute keeps a trunk revision of the module graph and says catalog-revision; one that moves no entry keeps + // none; a module leaving makes the next revision, naming it changed. + ix.Revisions = newMemObjects(0) + var revisions []meshcatalog.CatalogRevisionEvent + ix.SayCatalogRevision = func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error { + revisions = append(revisions, e) + return nil + } + if err := ix.Recompute(context.Background()); err != nil || len(revisions) != 1 || revisions[0].Revision != 1 || revisions[0].Kind != "trunk" || + revisions[0].Entries != int64(len(mods.vals)) || len(revisions[0].Changed) != len(mods.vals) { + t.Fatalf("the first recompute said %+v (%v), want trunk revision 1 with every module", revisions, err) + } graph := ix.Graph.(*memStore) written := graph.writes - if err := ix.Recompute(context.Background()); err != nil || graph.writes != written { - t.Errorf("a recompute that moves nothing wrote the graph %d times (%v)", graph.writes-written, err) + if err := ix.Recompute(context.Background()); err != nil || graph.writes != written || len(revisions) != 1 { + t.Errorf("a recompute that moves nothing wrote the graph %d times and said %d revisions (%v)", graph.writes-written, len(revisions), err) } // A module the modules state no longer holds (a merge dropped it, issue 1900) leaves the graph at the next recompute. delete(mods.vals, "other") if err := ix.Recompute(context.Background()); err != nil { t.Fatal(err) } + if len(revisions) != 2 || revisions[1].Revision != 2 || !reflect.DeepEqual(revisions[1].Changed, []string{"other"}) { + t.Errorf("after other left, said %+v, want revision 2 naming other changed", revisions) + } + if _, kept, err := ReadRevision(context.Background(), ix.Revisions, "trunk.2"); err != nil || len(kept) != len(mods.vals) { + t.Errorf("trunk.2 holds %d entries (%v), want %d", len(kept), err, len(mods.vals)) + } if _, held := graph.vals["other"]; held { t.Error("graph.other is still held after its module went") }