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 767449fe..788b6fa6 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -485,6 +485,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer pc.Stop() } + // The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the + // indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer + // starts again, said. + switch vcons, err := cred.EventConsumer(ctx, js, process.EventsStream); { + case err != nil: + logger.Printf("the consumer %s_%s on %s does not exist yet, so no package's series moves until the indexer starts again: %v", cred.Node, cred.Module, process.EventsStream, err) + default: + vc, err := process.HearEvents(vcons, func(m jetstream.Msg) { + if !heardVersion(ctx, ix, m, logger.Printf) { + _ = m.Ack() // nothing else the indexer consumes rides the shared stream + } + }, logger.Printf) + if err != nil { + return err + } + defer vc.Stop() + } cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { // Essential: the forge's merges are what the indexer indexes. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/series.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/series.go new file mode 100644 index 00000000..3d030bb1 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/series.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "encoding/json" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry" + "git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-indexer/internal/index" +) + +// The versions the packages state keeps per series (index.HearVersion, DLV-I5), heard on the indexer's durable consumer of +// the shared events stream: the package registry seats' published (their contracts in mesh-contracts) and the builder's +// numbered (its own event, version.numbered). +var ( + publishedSeats = map[string]string{ + gopackageregistry.SubjectPublished: gopackageregistry.Seat, + npmpackageregistry.SubjectPublished: npmpackageregistry.Seat, + } + numberedSubject = "mesh.mod." + delivery.BuilderModule + ".event.version.numbered" +) + +// numbered is the builder's version.numbered body as the indexer reads it. +type numbered struct { + Name string `json:"name"` + Kind string `json:"kind"` + Version string `json:"version"` + Tag string `json:"tag,omitempty"` + Fingerprint string `json:"fingerprint"` + Commit string `json:"commit,omitempty"` + Digest string `json:"digest,omitempty"` + Seat string `json:"seat,omitempty"` +} + +// heardVersion reads one message of the consumer as a version, false for any other subject. A package's is kept in its +// series; one that does not decode is said and acknowledged; a write that fails is left unacknowledged, so the bus gives +// it again. +func heardVersion(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) bool { + var h index.HeardVersion + switch seat, ok := publishedSeats[m.Subject()]; { + case ok: + var p gopackageregistry.PublishedEvent // both seats' published are the same Outcome + if err := json.Unmarshal(m.Data(), &p); err != nil { + logf("%s holds what is no published of %s, left: %v", m.Subject(), seat, err) + _ = m.Ack() + return true + } + if p.Kind != "package" { + _ = m.Ack() + return true + } + h = index.HeardVersion{Package: p.Name, Seat: seat, Version: p.Version, Tag: p.Tag, Commit: p.Commit, Digest: p.Digest, + Fingerprint: p.Fingerprint, GoModule: p.Module, Published: true} + case m.Subject() == numberedSubject: + var n numbered + if err := json.Unmarshal(m.Data(), &n); err != nil { + logf("%s holds what is no numbered, left: %v", m.Subject(), err) + _ = m.Ack() + return true + } + if n.Kind != "package" { + _ = m.Ack() + return true + } + h = index.HeardVersion{Package: n.Name, Seat: n.Seat, Version: n.Version, Tag: n.Tag, Commit: n.Commit, Digest: n.Digest, + Fingerprint: n.Fingerprint} + default: + return false + } + kept, err := ix.HearVersion(ctx, h) + if err != nil { + logf("%s of %s %s could not be kept in its series, so the bus gives it again: %v", m.Subject(), h.Package, h.Version, err) + return true + } + if kept { + logf("packages: %s %s kept in its series (published %v)", h.Package, h.Version, h.Published) + } + _ = m.Ack() + return true +} diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 4c66d058..610c5508 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -26,4 +26,9 @@ replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 +require ( + git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry v0.0.0-00010101000000-000000000000 +) + replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk diff --git a/modules/mesh-delivery-indexer/internal/index/control_records_test.go b/modules/mesh-delivery-indexer/internal/index/control_records_test.go index 55d50126..3102e381 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records_test.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records_test.go @@ -109,6 +109,35 @@ func TestTheControlRecords(t *testing.T) { t.Errorf("none = %#v, want an empty list", got) } }) + // DLV-I5 (milestone § 2 item 1, design 51 *The versions*): a package's series key keeps the newest trunk version the + // builder numbered and the newest a seat published; an older one, a tagged one and one of another series move neither; + // the per-module digests never read a series key. + t.Run("packages per series", func(t *testing.T) { + ctx := context.Background() + ix := &Indexer{Packages: newMem()} + sdkGo := "git.novox.be/novox/mesh-sdk/go" + for _, h := range []HeardVersion{ + {Package: "mesh-sdk/go", Seat: "go-package-registry", Version: "0.1.15", Commit: "c15"}, + {Package: "mesh-sdk/go", Seat: "go-package-registry", Version: "0.1.15", Commit: "c15", Digest: "h1:a", GoModule: sdkGo, Published: true}, + {Package: "mesh-sdk/go", Version: "0.1.16", Commit: "c16"}, + {Package: "mesh-sdk/go", Version: "0.1.14", Commit: "c14", Published: true}, + {Package: "mesh-sdk/go", Version: "0.1.17-feat-x.1", Tag: "feat-x", Published: true}, + {Package: "mesh-sdk/go", Version: "0.2.0", Commit: "c20", Published: true}, + } { + if _, err := ix.HearVersion(ctx, h); err != nil { + t.Fatal(err) + } + } + all := ix.SeriesAll(ctx) + one, two := all[SeriesKey("mesh-sdk/go", "0.1")], all[SeriesKey("mesh-sdk/go", "0.2")] + if one.Newest == nil || one.Newest.Version != "0.1.15" || one.Newest.Digest != "h1:a" || one.Numbered == nil || one.Numbered.Version != "0.1.16" || + one.GoModule != sdkGo || one.Seat != "go-package-registry" || len(all) != 2 || two.Newest == nil || two.Newest.Version != "0.2.0" { + t.Errorf("series = %+v", all) + } + if got := ix.publishedAll(ctx); len(got) != 0 { + t.Errorf("the per-module digests read a series key: %v", got) + } + }) ctx := context.Background() at := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC) t.Run("the predefined list, never replaced by an older merge", func(t *testing.T) { diff --git a/modules/mesh-delivery-indexer/internal/index/packages.go b/modules/mesh-delivery-indexer/internal/index/packages.go index de1f0e73..84a2ba7a 100644 --- a/modules/mesh-delivery-indexer/internal/index/packages.go +++ b/modules/mesh-delivery-indexer/internal/index/packages.go @@ -36,6 +36,9 @@ func (ix *Indexer) publishedAll(ctx context.Context) map[string]delivery.Publish return out } for _, key := range keys { + if IsSeriesKey(key) { + continue // a package's series (series.go), not a Go module's digests + } value, _, found, err := ix.Packages.Get(ctx, key) if err != nil || !found { continue diff --git a/modules/mesh-delivery-indexer/internal/index/series.go b/modules/mesh-delivery-indexer/internal/index/series.go new file mode 100644 index 00000000..84dc9bdf --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/series.go @@ -0,0 +1,142 @@ +package index + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The packages state per series (DLV-I5; milestone § 2 item 1; design 51 *The versions*, ADR 0606 §1, ADR 0609 §1): beside +// the per-module digests (packages.go, ADR 0593 §2), one key per package and series, `.`, holding the newest +// trunk version of that series the builder numbered and the newest a registry seat published. It is filled from the +// builder's `version.numbered` and the package registry seats' `published`, never from a timer; `indexer_requirements`, +// the fingerprint input and `indexer_package_users` read it. A tagged version (a group's) is never a series' newest. +// +// A series key always holds a dot (`mesh-sdk/go.0.1`), which EscapeName never leaves in a per-module key: a reader tells the +// two forms apart by IsSeriesKey. + +// SeriesPackage is the value of a series key. +type SeriesPackage struct { + V int `json:"v"` + // Package is the builder's name of the package, `/`; Series its `.`. + Package string `json:"package"` + Series string `json:"series"` + // Seat is the registry seat it is published through; GoModule the Go module path a Go package is fetched under (from + // go-package-registry's published), empty for any other. + Seat string `json:"seat,omitempty"` + GoModule string `json:"go-module,omitempty"` + // Newest is the newest version of the series a seat published: what a user accepting the series builds against. + Newest *SeriesVersion `json:"newest,omitempty"` + // Numbered is the newest version of the series the builder numbered, published or not yet. + Numbered *SeriesVersion `json:"numbered,omitempty"` + At time.Time `json:"at"` +} + +// SeriesVersion is one version of a series: its number, the source commit it was built from, the digest and fingerprint. +type SeriesVersion struct { + Version string `json:"version"` + Commit string `json:"commit,omitempty"` + Digest string `json:"digest,omitempty"` + Fingerprint string `json:"fingerprint,omitempty"` +} + +// SeriesKey is the key of a package's series in the packages state. +func SeriesKey(pkg, series string) string { return delivery.EscapeName(pkg) + "." + series } + +// IsSeriesKey says whether a key of the packages state is a series key, not a per-module one. +func IsSeriesKey(key string) bool { return strings.Contains(key, ".") } + +// SeriesOf is the series of a trunk version (`0.1.15` is of `0.1`); false for a version that is no `..` +// or that carries a tag (a pre-release). +func SeriesOf(version string) (string, bool) { + p, ok := semverParts("v" + strings.TrimPrefix(version, "v")) + if !ok || p.pre != "" { + return "", false + } + return fmt.Sprintf("%d.%d", p.core[0], p.core[1]), true +} + +// HeardVersion is what the indexer takes of a numbered or a published: the package's name, the seat, the version and its +// tag, the commit, digest and fingerprint, and the Go module path a published names. +type HeardVersion struct { + Package, Seat, Version, Tag, Commit, Digest, Fingerprint, GoModule string + // Published is true for a seat's published, false for the builder's numbered. + Published bool +} + +// HearVersion keeps a trunk version in its series' key: as the series' newest numbered, and, once published, as its newest. +// A tagged version, a version that does not read and an older one than held are passed over (false). Nil Packages keeps +// nothing. +func (ix *Indexer) HearVersion(ctx context.Context, h HeardVersion) (bool, error) { + series, ok := SeriesOf(h.Version) + if ix.Packages == nil || h.Package == "" || h.Tag != "" || !ok { + return false, nil + } + v := &SeriesVersion{Version: strings.TrimPrefix(h.Version, "v"), Commit: h.Commit, Digest: h.Digest, Fingerprint: h.Fingerprint} + newer := func(held *SeriesVersion) bool { + return held == nil || CompareVersions("v"+held.Version, "v"+v.Version) < 0 + } + kept := false + err := write(ctx, ix.Packages, SeriesKey(h.Package, series), func(value []byte, found bool) ([]byte, bool, error) { + s := SeriesPackage{V: delivery.Version, Package: h.Package, Series: series} + if found { + if err := json.Unmarshal(value, &s); err != nil { + ix.logf("packages.%s holds what is not a series, written anew: %v", SeriesKey(h.Package, series), err) + s = SeriesPackage{V: delivery.Version, Package: h.Package, Series: series} + } + } + kept = false + if newer(s.Numbered) { + s.Numbered, kept = v, true + } + if h.Published && newer(s.Newest) { + s.Newest, kept = v, true + } + if h.Seat != "" && s.Seat != h.Seat { + s.Seat, kept = h.Seat, true + } + if h.GoModule != "" && s.GoModule != h.GoModule { + s.GoModule, kept = h.GoModule, true + } + if !kept { + return nil, false, nil + } + s.At = ix.now() + next, err := json.Marshal(s) + return next, false, err + }) + return kept && err == nil, err +} + +// SeriesAll is every series key of the packages state, by key; empty with no state or one that does not answer. +func (ix *Indexer) SeriesAll(ctx context.Context) map[string]SeriesPackage { + out := map[string]SeriesPackage{} + if ix.Packages == nil { + return out + } + keys, err := ix.Packages.Keys(ctx) + if err != nil { + ix.logf("the packages state cannot be listed, so no series is read: %v", err) + return out + } + for _, key := range keys { + if !IsSeriesKey(key) { + continue + } + value, _, found, err := ix.Packages.Get(ctx, key) + if err != nil || !found { + continue + } + var s SeriesPackage + if err := json.Unmarshal(value, &s); err != nil { + ix.logf("packages.%s holds what is not a series, left out: %v", key, err) + continue + } + out[key] = s + } + return out +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 71cb3ef8..7e5f5ca9 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -123,7 +123,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control // internal/broker/state.go:62-93). - if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.placement", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + // The registry seats' published and the builder's numbered fill each package's series (DLV-I5, design 51 *The versions*). + if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.placement", + "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index e19baaf6..8c449c4a 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -40,11 +40,14 @@ "indexer_requirements" ], "consumes": [ + "go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.placement", + "mesh-delivery-builder.version.numbered", "mesh-delivery-checker.preview.written", - "mesh-delivery-forge-bridge.forge.**" + "mesh-delivery-forge-bridge.forge.**", + "npm-package-registry.published" ], "state": [ { "name": "candidates", "history": 10 },