DLV-I5: the indexer keeps each package's series, its newest numbered and newest published trunk version

A user takes the newest of the series it accepts (ADR 0609 §1), so the
indexer learns versions from the builder's numbered and the registry seats'
published, one key <package>.<series> beside the per-module digests.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 13:42:33 +01:00
parent ab502a7174
commit 1592abbf4f
8 changed files with 287 additions and 2 deletions
@@ -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.
@@ -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
}
+5
View File
@@ -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
@@ -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) {
@@ -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
@@ -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, `<package>.<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, `<module>/<artifact>`; Series its `<major>.<minor>`.
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 `<major>.<minor>.<patch>`
// 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
}
@@ -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" {
+4 -1
View File
@@ -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 },