Bridge, indexer and view: carry the catalogue's predefined list on its merged message, and keep it and each open pull request's config declarations in the indexer's states, so mesh-config reads them in place of the controller's mirror and pull-declares (issue 1953 D1-D3)
The view's tools read both states. Pins mesh-sdk 2d2aa7b.
This commit is contained in:
@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd h1:WhqMEXFZ9Zc+fEfLsnf1jhhGkc4IqMwUpXXwZq+V3+E=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd/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=
|
||||
|
||||
@@ -46,6 +46,13 @@ var ErrNotFound = errors.New("the forge has no such file")
|
||||
// says is not there is ErrNotFound, any other failure any other error (issue 1448).
|
||||
type Files func(owner, repo, path, ref string) ([]byte, error)
|
||||
|
||||
// CatalogRepository is the catalogue, whose merged message carries its predefined list, and PredefinedFile that list's
|
||||
// file at its root (novox/hq ADR 0565 §2, ADR 0348 §7).
|
||||
const (
|
||||
CatalogRepository = "novox/mesh-catalog"
|
||||
PredefinedFile = "predefined.json"
|
||||
)
|
||||
|
||||
// LockFiles are the lock files read beside a module's manifest, by name relative to its directory (novox/hq ADR 0465
|
||||
// clause 1: go.sum for a Go module, package-lock.json for a Node one).
|
||||
var LockFiles = []string{"go.sum", "package-lock.json"}
|
||||
@@ -315,6 +322,18 @@ func merged(subject string, p pull, trees Trees, files Files, kept Kept) (Messag
|
||||
}
|
||||
f := delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number,
|
||||
Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks}
|
||||
if repository == CatalogRepository {
|
||||
// The catalogue's predefined list at the merge (novox/hq ADR 0565 §2): the indexer keeps it for mesh-config.
|
||||
raw, err := files(p.Owner, p.Repo, PredefinedFile, p.MergeSHA)
|
||||
switch {
|
||||
case errors.Is(err, ErrNotFound):
|
||||
f.PredefinedRead = true
|
||||
case err != nil:
|
||||
return Message{}, fmt.Errorf("%w: %s: %s of %s at the merge commit %s: %v", ErrRetry, subject, PredefinedFile, repository, p.MergeSHA, err)
|
||||
default:
|
||||
f.Predefined = json.RawMessage(raw)
|
||||
}
|
||||
}
|
||||
body, err := delivery.EncodeForge(f)
|
||||
if err != nil {
|
||||
return Message{}, unreadable(subject, "%v", err)
|
||||
|
||||
@@ -97,6 +97,27 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1
|
||||
t.Errorf("the forge was asked for %s", c.path)
|
||||
}
|
||||
}
|
||||
// 1953/D1 (ADR 0565 §2): the catalogue's merged message carries its predefined.json at the merge, or says it holds
|
||||
// none; another repository's asks the forge for none.
|
||||
if !f.PredefinedRead || len(f.Predefined) != 0 {
|
||||
t.Errorf("a merge of the catalogue with no predefined.json does not say so: %s", msg.Data)
|
||||
}
|
||||
forge.have[PredefinedFile] = []byte(`{"font-size":{"type":"integer","value":13}}`)
|
||||
msg, _ = Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get)
|
||||
if f, err = delivery.DecodeForge(msg.Data); err != nil || string(f.Predefined) != string(forge.have[PredefinedFile]) || f.PredefinedRead {
|
||||
t.Errorf("predefined = %s (%v)", f.Predefined, err)
|
||||
}
|
||||
forge.calls = nil
|
||||
msg, _ = Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get)
|
||||
for _, c := range forge.calls {
|
||||
if c.path == PredefinedFile {
|
||||
t.Errorf("another repository's merge asked for %s", c.path)
|
||||
}
|
||||
}
|
||||
forge.errs = map[string]error{PredefinedFile: errors.New("forge down")}
|
||||
if _, err := Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get); !errors.Is(err, ErrRetry) {
|
||||
t.Errorf("a forge that cannot say is %v, want a retry", err)
|
||||
}
|
||||
// Issue 1126: a merge reads the forge only for the module directories whose tree changed.
|
||||
t.Run("a merge reads only the directories whose tree changed", mergeReadsOnlyChangedDirs)
|
||||
}
|
||||
|
||||
@@ -274,6 +274,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// 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)
|
||||
// The control records (ADR 0565 §2), new with this delivery and bound when they appear: until then nothing is written to them.
|
||||
predefined := process.AwaitBucket(ctx, js, conn, index.PredefinedBucket, "no predefined list is written", logger.Printf)
|
||||
configCandidates := process.AwaitBucket(ctx, js, conn, index.ConfigCandidatesBucket, "no pull request's config declarations are written", logger.Printf)
|
||||
// The controller's build records only name the directory of a module whose key names none (fillPaths): without them
|
||||
// such a module waits for its path, every other is indexed (issue 1945).
|
||||
builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf)
|
||||
@@ -303,7 +306,8 @@ 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}, Taken: awaitedStore{taken}, Builds: known, Logf: logger.Printf,
|
||||
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known,
|
||||
Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||
ReadOpen: openVia(calls.Call)}
|
||||
// The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a h1:nhRFmtdcI4VkghGUdJhpt1BnPuTQcekioOIRzW8fcG0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd h1:WhqMEXFZ9Zc+fEfLsnf1jhhGkc4IqMwUpXXwZq+V3+E=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd/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=
|
||||
|
||||
@@ -48,6 +48,8 @@ type Indexer struct {
|
||||
Modules Store
|
||||
// Manifests is the manifests bucket; nil writes no manifests (issue 1449).
|
||||
Manifests Store
|
||||
// Predefined and ConfigCandidates are the control records' states (control_records.go, ADR 0565 §2); nil writes none.
|
||||
Predefined, ConfigCandidates Store
|
||||
// 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.
|
||||
@@ -121,6 +123,14 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied,
|
||||
if err := ix.writeTakenKnown(ctx, f, w.asked); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
// The control records (ADR 0565 §2): the catalogue's predefined list, and the open pull requests' config
|
||||
// declarations judged again against the trunk this merge moved.
|
||||
if err := ix.writePredefined(ctx, f); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
if err := ix.configCandidatesAfterMerge(ctx, f, writtenModules(w.written)); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
}
|
||||
ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945)
|
||||
}
|
||||
@@ -134,6 +144,12 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied,
|
||||
if held, err = ix.heldModules(ctx); err != nil {
|
||||
return applied, err
|
||||
}
|
||||
if err := ix.configCandidateOfPull(ctx, f, held); err != nil {
|
||||
if !errors.Is(err, ErrUnreadable) {
|
||||
return applied, err
|
||||
}
|
||||
ix.logf("%s#%d: its config declarations are not written: %v", f.Repository, f.Number, err)
|
||||
}
|
||||
}
|
||||
if trunkLike[f.Branch] {
|
||||
ix.logf("%s#%d: its branch %q is no delivery group, so no candidate", f.Repository, f.Number, f.Branch)
|
||||
|
||||
@@ -0,0 +1,292 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// The indexer's two control records (novox/hq ADR 0565 §2, issue 1953 D2, D3): the catalogue's predefined list, read from the
|
||||
// forge bridge's merged message, and each open pull request's config declarations, read from its pull message's manifests
|
||||
// against the trunk's manifests the indexer holds. mesh-config reads both in place of the controller's mirror and its
|
||||
// `pull-declares`; a key's write is its event, nothing polls.
|
||||
|
||||
// The two states' buckets.
|
||||
var (
|
||||
PredefinedBucket = delivery.IndexerModule + "_" + delivery.IndexerPredefinedState
|
||||
ConfigCandidatesBucket = delivery.IndexerModule + "_" + delivery.IndexerConfigCandidatesState
|
||||
)
|
||||
|
||||
// writePredefined keeps the predefined list a merged message carries, unless the key names a later merge: a merge older
|
||||
// than the one it names never replaces it (ADR 0348 §7). The same list at the same merge writes nothing (ADR 0289 §1).
|
||||
func (ix *Indexer) writePredefined(ctx context.Context, f delivery.Forge) error {
|
||||
if ix.Predefined == nil || len(f.Predefined) == 0 {
|
||||
return nil
|
||||
}
|
||||
return ix.notYet(PredefinedBucket, write(ctx, ix.Predefined, delivery.PredefinedListKey, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
if found {
|
||||
var cur delivery.PredefinedKept
|
||||
if err := json.Unmarshal(value, &cur); err == nil {
|
||||
if cur.Commit == f.Merge && string(cur.List) == string(f.Predefined) {
|
||||
return nil, false, nil
|
||||
}
|
||||
older, err := ix.older(ctx, f.Repository, f.Merge, cur.Commit)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
if older {
|
||||
ix.logf("%s: the predefined list of %.12s is older than the one kept (%.12s): left", f.Repository, f.Merge, cur.Commit)
|
||||
return nil, false, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
next, err := json.Marshal(delivery.PredefinedKept{V: delivery.Version, Commit: f.Merge, At: ix.now(), List: f.Predefined})
|
||||
return next, false, err
|
||||
}))
|
||||
}
|
||||
|
||||
// notYet says a state not on the bus yet (it comes with this delivery, bound when it appears) and lets the message go:
|
||||
// what it would have written is written at the next message that carries it.
|
||||
func (ix *Indexer) notYet(bucket string, err error) error {
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
ix.logf("the state %s is not on the bus yet: nothing is written to it for this message", bucket)
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (ix *Indexer) now() time.Time {
|
||||
if ix.Now != nil {
|
||||
return ix.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
// PullName is a pull request as mesh-config names it, `owner/repository#number`.
|
||||
func PullName(repository string, number int) string { return repository + "#" + strconv.Itoa(number) }
|
||||
|
||||
// DeclarationsOf is a manifest's config declarations by key, in the layout of the controller's declared-config mirror: the
|
||||
// config form's `config`, else the settings form's `settings`, each default and enum encoded the one way.
|
||||
func DeclarationsOf(raw string) (map[string]delivery.ConfigDeclaration, error) {
|
||||
var m struct {
|
||||
Config map[string]delivery.ConfigDeclaration `json:"config"`
|
||||
Settings map[string]delivery.ConfigDeclaration `json:"settings"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(raw), &m); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
declared := m.Config
|
||||
if declared == nil {
|
||||
declared = m.Settings
|
||||
}
|
||||
out := map[string]delivery.ConfigDeclaration{}
|
||||
for key, d := range declared {
|
||||
for _, field := range []*json.RawMessage{&d.Default, &d.Enum} {
|
||||
if len(*field) == 0 {
|
||||
continue
|
||||
}
|
||||
var v any
|
||||
if err := json.Unmarshal(*field, &v); err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
canonical, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
*field = canonical
|
||||
}
|
||||
out[key] = d
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// changedDeclarations are the declarations head adds or changes against trunk.
|
||||
func changedDeclarations(head, trunk map[string]delivery.ConfigDeclaration) map[string]delivery.ConfigDeclaration {
|
||||
out := map[string]delivery.ConfigDeclaration{}
|
||||
for key, d := range head {
|
||||
if before, there := trunk[key]; !there || !reflect.DeepEqual(before, d) {
|
||||
out[key] = d
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// trunkManifest is the manifest the indexer holds for a module on the trunk, "" when it holds none.
|
||||
func (ix *Indexer) trunkManifest(ctx context.Context, held map[string]delivery.Module, module string) (string, error) {
|
||||
m, ok := held[module]
|
||||
if !ok || m.Manifest == "" || ix.Manifests == nil {
|
||||
return "", nil
|
||||
}
|
||||
raw, _, found, err := ix.Manifests.Get(ctx, m.Manifest)
|
||||
if err != nil || !found {
|
||||
return "", err
|
||||
}
|
||||
kept, err := delivery.DecodeManifest(raw)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return kept.JSON, nil
|
||||
}
|
||||
|
||||
// candidateDeclares are, per module of a pull request's head, the declarations its manifest adds or changes against the
|
||||
// trunk's (an empty set for a module that adds none): heads are the head's manifests by module.
|
||||
func (ix *Indexer) candidateDeclares(ctx context.Context, held map[string]delivery.Module, heads map[string]string) (map[string]map[string]delivery.ConfigDeclaration, error) {
|
||||
out := map[string]map[string]delivery.ConfigDeclaration{}
|
||||
for module, raw := range heads {
|
||||
head, err := DeclarationsOf(raw)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: the manifest of %s at the head: %v", ErrUnreadable, module, err)
|
||||
}
|
||||
trunkRaw, err := ix.trunkManifest(ctx, held, module)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
trunk := map[string]delivery.ConfigDeclaration{}
|
||||
if trunkRaw != "" {
|
||||
if trunk, err = DeclarationsOf(trunkRaw); err != nil {
|
||||
return nil, fmt.Errorf("the trunk's manifest of %s: %w", module, err)
|
||||
}
|
||||
}
|
||||
if changed := changedDeclarations(head, trunk); len(changed) > 0 {
|
||||
out[module] = changed
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// writeConfigCandidate writes a pull request's key of config-candidates from its head's manifests by module: written when
|
||||
// what it says differs (its time apart), never for the same head and declarations twice.
|
||||
func (ix *Indexer) writeConfigCandidate(ctx context.Context, repository string, number int, head string, held map[string]delivery.Module, heads map[string]string) error {
|
||||
if ix.ConfigCandidates == nil {
|
||||
return nil
|
||||
}
|
||||
pull := PullName(repository, number)
|
||||
key, err := delivery.ConfigCandidateKey(pull)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %v", ErrUnreadable, err)
|
||||
}
|
||||
declares, err := ix.candidateDeclares(ctx, held, heads)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return ix.notYet(ConfigCandidatesBucket, write(ctx, ix.ConfigCandidates, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
if found {
|
||||
var cur delivery.ConfigCandidate
|
||||
if json.Unmarshal(value, &cur) == nil && cur.Head == head && reflect.DeepEqual(cur.Declares, declares) {
|
||||
return nil, false, nil
|
||||
}
|
||||
}
|
||||
next, err := json.Marshal(delivery.ConfigCandidate{V: delivery.Version, Pull: pull, Head: head, Declares: declares, At: ix.now()})
|
||||
return next, false, err
|
||||
}))
|
||||
}
|
||||
|
||||
// dropConfigCandidate deletes a pull request's key: it merged or closed.
|
||||
func (ix *Indexer) dropConfigCandidate(ctx context.Context, repository string, number int) error {
|
||||
if ix.ConfigCandidates == nil {
|
||||
return nil
|
||||
}
|
||||
key, err := delivery.ConfigCandidateKey(PullName(repository, number))
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return ix.notYet(ConfigCandidatesBucket, write(ctx, ix.ConfigCandidates, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }))
|
||||
}
|
||||
|
||||
// configCandidateOfPull is a pull message's part in config-candidates: a head whose manifests the message carries is written,
|
||||
// a closed one deleted. A head held again without its directories leaves the key as it is.
|
||||
func (ix *Indexer) configCandidateOfPull(ctx context.Context, f delivery.Forge, held map[string]delivery.Module) error {
|
||||
if f.Action == delivery.PullClosed {
|
||||
return ix.dropConfigCandidate(ctx, f.Repository, f.Number)
|
||||
}
|
||||
if !f.DirsKnown {
|
||||
return nil
|
||||
}
|
||||
heads := map[string]string{}
|
||||
for dir, raw := range f.Manifests {
|
||||
heads[moduleOfDir(held, f, dir)] = raw
|
||||
}
|
||||
return ix.writeConfigCandidate(ctx, f.Repository, f.Number, f.Head, held, heads)
|
||||
}
|
||||
|
||||
// configCandidatesAfterMerge writes again every open pull request's key that names a module the merge changed, since what
|
||||
// it adds is judged against the trunk (ADR 0565 §2), from the head's manifests its candidate keeps; and deletes the merged
|
||||
// pull request's.
|
||||
func (ix *Indexer) configCandidatesAfterMerge(ctx context.Context, f delivery.Forge, changed []string) error {
|
||||
if ix.ConfigCandidates == nil {
|
||||
return nil
|
||||
}
|
||||
if err := ix.dropConfigCandidate(ctx, f.Repository, f.Number); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(changed) == 0 || ix.Store == nil || ix.Manifests == nil {
|
||||
return nil
|
||||
}
|
||||
held, err := ix.heldModules(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := ix.ConfigCandidates.Keys(ctx); errors.Is(err, process.ErrNotYet) {
|
||||
return nil
|
||||
}
|
||||
groups, err := ix.Store.Keys(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var errs []error
|
||||
for _, g := range groups {
|
||||
raw, _, found, err := ix.Store.Get(ctx, g)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
c, err := delivery.DecodeCandidate(raw)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, p := range c.Pulls {
|
||||
if !touchesAny(p, changed) {
|
||||
continue
|
||||
}
|
||||
heads := map[string]string{}
|
||||
for module, fp := range p.Fps {
|
||||
mraw, _, found, err := ix.Manifests.Get(ctx, fp.Manifest)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
if kept, err := delivery.DecodeManifest(mraw); err == nil {
|
||||
heads[module] = kept.JSON
|
||||
}
|
||||
}
|
||||
if err := ix.writeConfigCandidate(ctx, p.Repository, p.Number, p.Head.Head, held, heads); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
func touchesAny(p delivery.CandidatePull, modules []string) bool {
|
||||
for _, m := range modules {
|
||||
if _, ok := p.Fps[m]; ok {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// writtenModules are the modules a merged message wrote, from its written keys (`modules.<module>`).
|
||||
func writtenModules(written []string) []string {
|
||||
var out []string
|
||||
for _, k := range written {
|
||||
out = append(out, strings.TrimPrefix(k, "modules."))
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// 1953/D2, D3 (novox/hq ADR 0565 §2): the indexer keeps the catalogue's predefined list from the merged message, a merge older
|
||||
// than the one kept never replacing it; and each open pull request's config declarations its head adds or changes against the
|
||||
// trunk's manifest, an empty set when none, written only when they change and deleted when it merges or closes.
|
||||
func TestTheControlRecords(t *testing.T) {
|
||||
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) {
|
||||
ix, _, _ := manifestsIndexer(t)
|
||||
pre := newMem()
|
||||
ix.Predefined, ix.Now = pre, func() time.Time { return at }
|
||||
f := mergedMsg(catalogueRepo, map[string]string{"modules/a": "t1"})
|
||||
f.Predefined = json.RawMessage(`{"font-size":{"type":"integer","value":13}}`)
|
||||
if err := ix.Apply(ctx, f); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var kept delivery.PredefinedKept
|
||||
if err := json.Unmarshal(pre.vals[delivery.PredefinedListKey], &kept); err != nil || kept.Commit != merge || string(kept.List) != string(f.Predefined) {
|
||||
t.Fatalf("kept = %+v (%v)", kept, err)
|
||||
}
|
||||
writes := pre.writes
|
||||
if err := ix.Apply(ctx, f); err != nil || pre.writes != writes {
|
||||
t.Errorf("the same list at the same merge wrote again (%v)", err)
|
||||
}
|
||||
older := f
|
||||
older.Merge, older.Predefined = strings.Repeat("e", 40), json.RawMessage(`{}`)
|
||||
ix.Contains = func(context.Context, string, string, string) (bool, error) { return true, nil }
|
||||
if err := ix.writePredefined(ctx, older); err != nil || pre.writes != writes {
|
||||
t.Errorf("an older merge replaced the list (%v)", err)
|
||||
}
|
||||
ix.Predefined = &memStore{unbound: process.ErrNotYet}
|
||||
if err := ix.writePredefined(ctx, f); err != nil {
|
||||
t.Errorf("a state not on the bus yet stopped the message: %v", err)
|
||||
}
|
||||
})
|
||||
t.Run("a pull request's config declarations against the trunk", func(t *testing.T) {
|
||||
ix, _, _ := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
cands := newMem()
|
||||
ix.ConfigCandidates, ix.Now = cands, func() time.Time { return at }
|
||||
trunk := `{"module":"avahi","version":"1","config":{"port":{"type":"integer","default":5353,"why":"mDNS"}}}`
|
||||
if err := ix.Apply(ctx, withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "t1"}), map[string]string{"modules/avahi": trunk}, nil)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key := "novox.mesh-catalog.7"
|
||||
head := strings.Replace(trunk, `"why":"mDNS"}`, `"why":"mDNS"},"name":{"type":"string","default":"x","why":"shown"}`, 1)
|
||||
p := withManifests(pull(catalogueRepo, 7, "feat/y", h1, "modules/avahi"), map[string]string{"modules/avahi": head}, nil)
|
||||
if err := ix.Apply(ctx, p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var c delivery.ConfigCandidate
|
||||
if err := json.Unmarshal(cands.vals[key], &c); err != nil || c.Pull != "novox/mesh-catalog#7" || c.Head != h1 || len(c.Declares["avahi"]) != 1 ||
|
||||
string(c.Declares["avahi"]["name"].Default) != `"x"` {
|
||||
t.Fatalf("candidate = %+v (%v)", c, err)
|
||||
}
|
||||
writes := cands.writes
|
||||
if err := ix.Apply(ctx, p); err != nil || cands.writes != writes {
|
||||
t.Errorf("the same head wrote again (%v)", err)
|
||||
}
|
||||
same := withManifests(pull(catalogueRepo, 7, "feat/y", strings.Repeat("d", 40), "modules/avahi"), map[string]string{"modules/avahi": trunk}, nil)
|
||||
if err := ix.Apply(ctx, same); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c = delivery.ConfigCandidate{}
|
||||
if err := json.Unmarshal(cands.vals[key], &c); err != nil || len(c.Declares) != 0 || c.Declares == nil {
|
||||
t.Errorf("a head declaring nothing new = %+v, want an empty set", c)
|
||||
}
|
||||
closed := pull(catalogueRepo, 7, "feat/y", strings.Repeat("d", 40))
|
||||
closed.Action = delivery.PullClosed
|
||||
if err := ix.Apply(ctx, closed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := cands.vals[key]; ok {
|
||||
t.Error("a closed pull request's declarations stayed")
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -42,7 +42,12 @@ func withManifests(f delivery.Forge, manifests map[string]string, locks map[stri
|
||||
return f
|
||||
}
|
||||
|
||||
func TestAMergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1
|
||||
func TestAMergedManifestIsKeptOnceByItsDigestAndNamedByTheModule(t *testing.T) {
|
||||
t.Run("kept by its digest and named by the module", mergedManifestIsKeptByItsDigestAndNamedByTheModule)
|
||||
t.Run("the same manifest applied again is written once", sameManifestAppliedAgainIsWrittenOnce)
|
||||
}
|
||||
|
||||
func mergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1
|
||||
avahi := catalogueManifest(t, "avahi")
|
||||
ix, mods, manifests := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "newtree"}), map[string]string{"modules/avahi": avahi}, nil)
|
||||
@@ -63,7 +68,7 @@ func TestAMergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { //
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheSameManifestAppliedAgainIsWrittenOnce(t *testing.T) { // 1449/AC2
|
||||
func sameManifestAppliedAgainIsWrittenOnce(t *testing.T) { // 1449/AC2
|
||||
avahi := catalogueManifest(t, "avahi")
|
||||
ix, _, manifests := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "newtree"}), map[string]string{"modules/avahi": avahi}, nil)
|
||||
|
||||
@@ -86,7 +86,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
t.Errorf("emits = %v, want none", m.Emits)
|
||||
}
|
||||
// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) {
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {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.builds", "mesh-delivery-assigner.placement", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
|
||||
@@ -16,6 +16,8 @@
|
||||
{ "name": "heard", "history": 64 },
|
||||
{ "name": "manifests", "history": 1 },
|
||||
{ "name": "modules", "history": 20 },
|
||||
{ "name": "predefined", "history": 5 },
|
||||
{ "name": "config-candidates", "history": 5 },
|
||||
{ "name": "taken", "history": 1 },
|
||||
{ "name": "takeins", "history": 1 }
|
||||
],
|
||||
|
||||
@@ -104,6 +104,23 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
|
||||
group, _ := args["group"].(string)
|
||||
return r.Candidate(group)
|
||||
}},
|
||||
{Name: view.PredefinedReadTool,
|
||||
Description: "The catalogue's predefined list as the indexer keeps it (novox/hq ADR 0565 §2): its key list, the list whole as " +
|
||||
"predefined.json holds it with the merge it was read at; a merge older than that never replaces it. mesh-config " +
|
||||
"publishes the predefined keys from it after its take-over. Read-only.",
|
||||
Run: func(map[string]any) (any, error) {
|
||||
return view.ControlRecord(r.PredefinedState, delivery.IndexerPredefinedState, delivery.PredefinedListKey)
|
||||
}},
|
||||
{Name: view.ConfigCandidatesReadTool,
|
||||
Description: "Each open pull request's config declarations as the indexer keeps them (novox/hq ADR 0565 §2): per pull " +
|
||||
"request (owner.repository.number), its head and per module the config keys its manifests add or change against the " +
|
||||
"trunk's, an empty set when none; deleted when it merges or closes. Give key for one. Read-only.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{
|
||||
"key": map[string]any{"type": "string", "description": "one pull request's key, owner.repository.number"}}},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
key, _ := args["key"].(string)
|
||||
return view.ControlRecord(r.ConfigCandidatesState, delivery.IndexerConfigCandidatesState, key)
|
||||
}},
|
||||
{Name: view.CandidatesReadTool,
|
||||
Description: "Every candidate the indexer holds, by group, each as stored under candidates.<group>; with module, only those with a " +
|
||||
"pull request touching that module. Keys that cannot be read are listed as unreadable. Read-only.",
|
||||
@@ -245,7 +262,9 @@ func main() {
|
||||
HeardState: read(stdio.State("mesh-delivery-indexer.heard")),
|
||||
SignersState: read(stdio.State("mesh-vault.signers")),
|
||||
ConditionsState: read(stdio.State(view.StatusConditionsState)), AttemptsState: read(stdio.State(view.StatusAttemptsState)),
|
||||
PresenceState: read(stdio.State(view.StatusPresenceState)), StuckState: read(stdio.State(view.StatusStuckState))}
|
||||
PresenceState: read(stdio.State(view.StatusPresenceState)), StuckState: read(stdio.State(view.StatusStuckState)),
|
||||
PredefinedState: read(stdio.State(delivery.IndexerModule + "." + delivery.IndexerPredefinedState)),
|
||||
ConfigCandidatesState: read(stdio.State(delivery.IndexerModule + "." + delivery.IndexerConfigCandidatesState))}
|
||||
u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")),
|
||||
Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
|
||||
for _, lang := range delivery.AskLanguages {
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-view
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd h1:WhqMEXFZ9Zc+fEfLsnf1jhhGkc4IqMwUpXXwZq+V3+E=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd/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=
|
||||
|
||||
@@ -50,6 +50,41 @@ type Reader struct {
|
||||
// ConditionsState, AttemptsState, PresenceState and StuckState are what status reads beside the wants and the haves
|
||||
// (ADR 0564 §10): the condition keeper's open conditions, mesh-healers' attempts, the watchman's presence and stuck keys.
|
||||
ConditionsState, AttemptsState, PresenceState, StuckState StateRead
|
||||
// PredefinedState and ConfigCandidatesState are the indexer's control records (ADR 0565 §2, §15); unset reads none.
|
||||
PredefinedState, ConfigCandidatesState StateRead
|
||||
}
|
||||
|
||||
// The read tools of the indexer's control records (ADR 0565 §15): the indexer serves no tool of its own (ADR 0437 §4).
|
||||
const (
|
||||
PredefinedReadTool = "indexer_predefined"
|
||||
ConfigCandidatesReadTool = "indexer_config_candidates"
|
||||
)
|
||||
|
||||
// ControlRecord is every key of one of the indexer's control records as stored, or one key, with how many it holds.
|
||||
func ControlRecord(s StateRead, name, key string) (any, error) {
|
||||
if s.Keys == nil || s.Get == nil {
|
||||
return nil, fmt.Errorf("the indexer's %s cannot be read here", name)
|
||||
}
|
||||
if key = strings.TrimSpace(key); key != "" {
|
||||
raw, err := s.Get(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if raw == nil {
|
||||
return nil, fmt.Errorf("the indexer's %s holds no key %s", name, key)
|
||||
}
|
||||
return raw, nil
|
||||
}
|
||||
keys, err := s.Keys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sort.Strings(keys)
|
||||
values, err := s.values(keys)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"state": name, "holds": len(keys), "keys": values}, nil
|
||||
}
|
||||
|
||||
// Status is the view's derivation of status at now (status.go).
|
||||
|
||||
@@ -87,7 +87,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-conditions.conditions", "mesh-config.resolved", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.placement", "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.heard", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-vault.signers", "node-engine.have"} // view_status reads the keeper's conditions and the healers' attempts (1952/V1); 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-conditions.conditions", "mesh-config.resolved", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.placement", "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.config-candidates", "mesh-delivery-indexer.heard", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.predefined", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-vault.signers", "node-engine.have"} // view_status reads the keeper's conditions and the healers' attempts (1952/V1); 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) {
|
||||
@@ -119,7 +119,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{"view_status", "indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "node_trusts", "build_usage", "view_reach", "indexer_module_history", "indexer_heard"}; !reflect.DeepEqual(m.Tools, want) {
|
||||
if want := []string{"view_status", "indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "indexer_predefined", "indexer_config_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "node_trusts", "build_usage", "view_reach", "indexer_module_history", "indexer_heard"}; !reflect.DeepEqual(m.Tools, want) {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||
}
|
||||
for _, a := range m.Build.Artifacts {
|
||||
|
||||
@@ -9,6 +9,8 @@
|
||||
"indexer_modules",
|
||||
"indexer_candidate",
|
||||
"indexer_candidates",
|
||||
"indexer_predefined",
|
||||
"indexer_config_candidates",
|
||||
"builder_build",
|
||||
"builder_builds",
|
||||
"assigner_wants",
|
||||
@@ -50,8 +52,10 @@
|
||||
"mesh-delivery-gate.asks",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.config-candidates",
|
||||
"mesh-delivery-indexer.heard",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.predefined",
|
||||
"mesh-delivery-indexer.taken",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-conditions.conditions",
|
||||
|
||||
Reference in New Issue
Block a user