Merge commit 'ef13e63' into feat/take-in
This commit is contained in:
@@ -8,6 +8,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
@@ -191,8 +192,32 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// keptRead is a state the tools bundle reads, as the index package reads a store: Get and Keys; it writes nothing.
|
||||
type keptRead struct{ s stdio.KeptState }
|
||||
|
||||
func (k keptRead) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
|
||||
e, err := k.s.Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, 0, false, err
|
||||
}
|
||||
return e.Value, e.Revision, true, nil
|
||||
}
|
||||
|
||||
func (k keptRead) Keys(context.Context) ([]string, error) { return k.s.Keys() }
|
||||
|
||||
func (keptRead) Create(context.Context, string, []byte) error { return errReadOnly }
|
||||
|
||||
func (keptRead) Update(context.Context, string, []byte, uint64) error { return errReadOnly }
|
||||
|
||||
func (keptRead) Delete(context.Context, string, uint64) error { return errReadOnly }
|
||||
|
||||
var errReadOnly = errors.New("the tools bundle only reads this state")
|
||||
|
||||
func main() {
|
||||
if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)); err != nil {
|
||||
ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second)
|
||||
tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)
|
||||
tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...)
|
||||
if err := stdio.Serve("", tools); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
@@ -9,7 +9,9 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -20,6 +22,10 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/graph"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
|
||||
@@ -42,11 +48,58 @@ accounts {
|
||||
// bus with the stream FORGE and the indexer's durable <node>_<module> on it, the buckets the controller makes from the
|
||||
// manifest's state (mesh-delivery-indexer_candidates, history 10) and mesh-controller_builds holding build records, and
|
||||
// no mesh-delivery-gate_source bucket.
|
||||
func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
|
||||
js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds buildLog) {
|
||||
//
|
||||
// And the take-in step (ADR 0633, issue 1585 I8a, I8b): the indexer's states pending, verdicts, graph, taken and packages
|
||||
// and its stream take-in are there, mesh-build-go reported its toolchain, and the node-engine's composition answers the
|
||||
// build of the indexer that runs. A merge changing the indexer's own tree writes only the indexer's entry, its manifest,
|
||||
// its verdict and its key of the graph (and the packages it publishes); the rest is pending, said take-in.pending and
|
||||
// named by indexer_pending with the build it waits for. A group's first member is gathered meanwhile. Started again as
|
||||
// that build, the indexer takes the revision in and leaves the group gathered; the group's second member and its end
|
||||
// then make one take-in, said take-in.taken under the group's key.
|
||||
func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
|
||||
var running atomic.Pointer[string]
|
||||
first := "fp-the-indexer-that-runs"
|
||||
running.Store(&first)
|
||||
js, ctx, candidates, _, restart := startIndexerRun(t, func(ctx context.Context, builds buildLog) {
|
||||
// The builder's log entries of two builds' ends (ADR 0571 clause 1), as the builder appends them.
|
||||
builds.end(t, "novox/mesh-delivery", "a", "modules/a")
|
||||
builds.end(t, "novox/mesh-catalog", "c", "modules/c")
|
||||
}, func(ctx context.Context, js jetstream.JetStream) {
|
||||
// The states and the stream the controller makes from the indexer's manifest (old sections, ADR 0595 §11).
|
||||
for bucket, history := range map[string]uint8{index.PendingBucket: 1, delivery.VerdictsBucket: delivery.VerdictsHistory,
|
||||
graph.Bucket: 1, delivery.TakenBucket: 1, delivery.PackagesBucket: 1} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: history}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for name, subject := range map[string]string{"TAKE_IN": "mesh.stream.mesh-delivery-indexer.event.take-in.>",
|
||||
"INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subject}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
tc, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := delivery.ToolchainKey("n1", "go")
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "1", At: time.Now()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := tc.Put(ctx, key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The node-engine's composition on n1 (process.Conn.Call's subjects): the build of the indexer that runs there.
|
||||
answer := func(m *nats.Msg) {
|
||||
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"runs": map[string]string{"fp": *running.Load()}}, "node": "n1"})
|
||||
_ = m.Respond(raw)
|
||||
}
|
||||
for _, subject := range []string{"mesh.seat.node-engine.call.composition.n1.>", "mesh.seat.node-engine.tool.composition.n1"} {
|
||||
if _, err := js.Conn().Subscribe(subject, answer); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
})
|
||||
h1, h2 := "0123456789abcdef0123456789abcdef01234567", "89abcdef0123456789abcdef0123456789abcdef"
|
||||
for _, f := range []delivery.Forge{
|
||||
@@ -83,11 +136,186 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
|
||||
t.Errorf("%s#%d modules = %v, want [%s]", p.Repository, p.Number, p.Modules, want)
|
||||
}
|
||||
}
|
||||
|
||||
// I8a: pull request 12 merged, changing the indexer's own tree and a's.
|
||||
manifest, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const ix = delivery.IndexerModule
|
||||
m1 := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: h1,
|
||||
Merge: "1111111111111111111111111111111111111111", Root: "r0",
|
||||
Trees: map[string]string{"modules/" + ix: "tree-indexer-2", "modules/a": "tree-a-2"},
|
||||
Manifests: map[string]string{"modules/" + ix: string(manifest), "modules/a": `{"module":"a","version":"1"}`},
|
||||
Locks: map[string]map[string]string{"modules/" + ix: {"go.sum": "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"}}}
|
||||
merged := func(f delivery.Forge) uint64 {
|
||||
t.Helper()
|
||||
data, err := delivery.EncodeForge(f)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
event, _ := delivery.ForgeEvent(f.Repository, delivery.ForgeMerged)
|
||||
ack, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event."+event, data)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ack.Sequence
|
||||
}
|
||||
m1Key := delivery.PendingKey(merged(m1))
|
||||
takeIns, err := js.Stream(ctx, "TAKE_IN")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// said is the last take-in event under kind and key, within 20 s.
|
||||
said := func(kind, key string) (delivery.TakeIn, bool) {
|
||||
event, _ := delivery.TakeInEvent(kind, key)
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if msg, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, event)); err == nil {
|
||||
if ti, err := delivery.DecodeTakeIn(msg.Data); err == nil {
|
||||
return ti, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return delivery.TakeIn{}, false
|
||||
}
|
||||
held, ok := said(delivery.TakeInPending, m1Key)
|
||||
if !ok || held.What != delivery.TakeInRevision || held.WaitsFor == "" || held.WaitsFor == first || len(held.Merges) != 1 {
|
||||
t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the revision waiting for another build than %s", m1Key, held, ok, first)
|
||||
}
|
||||
named := held.WaitsFor
|
||||
get := func(bucket, key string) ([]byte, bool) {
|
||||
t.Helper()
|
||||
kv, err := js.KeyValue(ctx, bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
e, err := kv.Get(ctx, key)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
return e.Value(), true
|
||||
}
|
||||
// Only what building and moving the named indexer needs is written.
|
||||
raw, ok := get(delivery.ModulesBucket, ix)
|
||||
own, err := delivery.DecodeModule(raw)
|
||||
if !ok || err != nil || own.FP2 != named || own.Commit != m1.Merge {
|
||||
t.Fatalf("modules.%s = %+v (%v), want the named build %s at the merge", ix, own, err, named)
|
||||
}
|
||||
if _, ok := get(delivery.ManifestsBucket, own.Manifest); !ok {
|
||||
t.Errorf("manifests.%s is not written", own.Manifest)
|
||||
}
|
||||
if _, ok := get(delivery.VerdictsBucket, own.Manifest); !ok {
|
||||
t.Errorf("the verdict on the indexer's manifest %s is not written", own.Manifest)
|
||||
}
|
||||
if gk, _ := graph.Key(ix); !func() bool { _, ok := get(graph.Bucket, gk); return ok }() {
|
||||
t.Errorf("the indexer's key %s of the module graph is not written", gk)
|
||||
}
|
||||
repoKey, _ := delivery.RepositoryKey(m1.Repository)
|
||||
if _, ok := get(delivery.ModulesBucket, "a"); ok {
|
||||
t.Errorf("modules.a is written while the revision is pending")
|
||||
}
|
||||
if _, ok := get(delivery.TakenBucket, repoKey); ok {
|
||||
t.Errorf("taken.%s is written while the revision is pending", repoKey)
|
||||
}
|
||||
if e, err := candidates.Get(ctx, "feat/x"); err != nil {
|
||||
t.Errorf("the candidate feat/x is gone while the merge of #12 is pending: %v", err)
|
||||
} else if c, err := delivery.DecodeCandidate(e.Value()); err != nil || len(c.Pulls) != 2 {
|
||||
t.Errorf("the candidate feat/x = %+v while the merge of #12 is pending, want its two pull requests", c)
|
||||
}
|
||||
|
||||
// A group's first member arrives meanwhile; its end is still to come (the knowledge base's remark: it stays across the restart).
|
||||
members := []delivery.Head{{Repository: "novox/mesh-catalog", Number: 40, Head: "4040404040404040404040404040404040404040"},
|
||||
{Repository: "novox/mesh-controller", Number: 41, Head: "4141414141414141414141414141414141414141"}}
|
||||
g1 := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/g", Number: 40, Head: members[0].Head,
|
||||
Merge: "4000000000000000000000000000000000000000", Root: "r0", Trees: map[string]string{"modules/c": "tree-c-2"},
|
||||
Manifests: map[string]string{"modules/c": `{"module":"c","version":"1"}`}, Group: "feat/g", Members: members}
|
||||
g2 := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-controller", Branch: "feat/g", Number: 41, Head: members[1].Head,
|
||||
Merge: "4100000000000000000000000000000000000000", Root: "r0", Trees: map[string]string{"modules/k": "tree-k-1"},
|
||||
Manifests: map[string]string{"modules/k": `{"module":"k","version":"1"}`}, Group: "feat/g", Members: members}
|
||||
g1Key, groupKey := delivery.PendingKey(merged(g1)), delivery.GroupTakeInKey("feat/g")
|
||||
if gathered, ok := said(delivery.TakeInPending, groupKey); !ok || gathered.What != delivery.TakeInGroup || len(gathered.Merges) != 1 {
|
||||
t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the group with its first member", groupKey, gathered, ok)
|
||||
}
|
||||
// indexer_pending, the tool over the same state, names both waits.
|
||||
pendingKV, err := js.KeyValue(ctx, index.PendingBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waits := func() []delivery.PendingWait {
|
||||
t.Helper()
|
||||
answer, err := index.PendingTools(kvStore{pendingKV}, nil, time.Now)[0].Run(map[string]any{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return answer.(delivery.PendingAnswer).Waits
|
||||
}
|
||||
var w []delivery.PendingWait
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if w = waits(); len(w) == 2 && w[0].WaitsFor == named && w[1].Said != "" {
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(w) != 2 || w[0].Key != m1Key || w[0].WaitsFor != named || !slices.Contains(w[0].Holds, "a") || !slices.Contains(w[0].Holds, ix) ||
|
||||
w[1].Key != groupKey || w[1].What != delivery.TakeInGroup || len(w[1].Merges) != 1 || !strings.HasSuffix(w[1].Said, "pending."+groupKey) {
|
||||
t.Fatalf("indexer_pending = %+v, want the revision %s waiting for %s holding a and the indexer, then the group gathered", w, m1Key, named)
|
||||
}
|
||||
|
||||
// The named indexer starts (the node-engine now runs that build): it takes the revision in, and leaves the group.
|
||||
running.Store(&named)
|
||||
restart()
|
||||
if taken, ok := said(delivery.TakeInTaken, m1Key); !ok || len(taken.Merges) != 1 || taken.Merges[0].Merge != m1.Merge {
|
||||
t.Fatalf("take-in.taken.%s = %+v (said %v) within 20 s of the start of %s", m1Key, taken, ok, named)
|
||||
}
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if w = waits(); len(w) == 1 {
|
||||
break
|
||||
}
|
||||
}
|
||||
if raw, ok := get(delivery.ModulesBucket, "a"); !ok || !strings.Contains(string(raw), m1.Merge) {
|
||||
t.Errorf("modules.a = %s after the take-in at the start, want it at the merge %s", raw, m1.Merge)
|
||||
}
|
||||
if _, ok := get(delivery.TakenBucket, repoKey); !ok {
|
||||
t.Errorf("taken.%s is not written by the take-in at the start", repoKey)
|
||||
}
|
||||
if len(w) != 1 || w[0].Key != groupKey || len(w[0].Keys) != 1 || w[0].Keys[0] != g1Key {
|
||||
t.Fatalf("indexer_pending after the start = %+v, want only the group gathered, its first member %s", w, g1Key)
|
||||
}
|
||||
if _, ok := get(delivery.ModulesBucket, "c"); ok {
|
||||
t.Errorf("modules.c is written while its group is gathered")
|
||||
}
|
||||
|
||||
// I8b: the group's second member and its end: one take-in of both, said under the group's key.
|
||||
merged(g2)
|
||||
end, err := delivery.EncodeForgeGroupEnd(delivery.ForgeGroupEnd{Group: "feat/g", Members: []delivery.GroupMerge{
|
||||
{Repository: g1.Repository, Number: g1.Number, Merge: g1.Merge}, {Repository: g2.Repository, Number: g2.Number, Merge: g2.Merge}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event."+delivery.ForgeGroupEndEvent, end); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if taken, ok := said(delivery.TakeInTaken, groupKey); !ok || taken.What != delivery.TakeInGroup || len(taken.Merges) != 2 {
|
||||
t.Fatalf("take-in.taken.%s = %+v (said %v) within 20 s of the group's end, want both members", groupKey, taken, ok)
|
||||
}
|
||||
var hasC, hasK bool
|
||||
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && !(hasC && hasK && len(w) == 0); time.Sleep(50 * time.Millisecond) {
|
||||
_, hasC = get(delivery.ModulesBucket, "c")
|
||||
_, hasK = get(delivery.ModulesBucket, "k")
|
||||
w = waits()
|
||||
}
|
||||
if !hasC || !hasK || len(w) != 0 {
|
||||
t.Errorf("after the group's end: modules.c %v, modules.k %v, indexer_pending %+v: want both taken in and nothing waiting", hasC, hasK, w)
|
||||
}
|
||||
// One take-in: neither member was said as a revision of its own.
|
||||
single, _ := delivery.TakeInEvent(delivery.TakeInPending, g1Key)
|
||||
if _, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, single)); err == nil {
|
||||
t.Errorf("the group's first member %s was said pending as a revision of its own", g1Key)
|
||||
}
|
||||
}
|
||||
|
||||
// startIndexer starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill
|
||||
// mesh-controller_builds, and runs the program. mesh-delivery-gate_source is never created.
|
||||
func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) {
|
||||
// startIndexerRun starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill
|
||||
// mesh-controller_builds, and runs the program; restart starts it again. mesh-delivery-gate_source is never created.
|
||||
func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue, func()) {
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -166,11 +394,24 @@ func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...f
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
var stop func()
|
||||
start := func() {
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
stop = func() { cancel(); <-done }
|
||||
}
|
||||
start()
|
||||
t.Cleanup(func() { stop() })
|
||||
// restart stops the program and starts it again on the same bus, as the node-engine does with a new build.
|
||||
restart := func() { stop(); start() }
|
||||
|
||||
return js, ctx, candidates, modules, restart
|
||||
}
|
||||
|
||||
// startIndexer is startIndexerRun for a test that never starts the program again.
|
||||
func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) {
|
||||
js, ctx, candidates, modules, _ := startIndexerRun(t, put, before...)
|
||||
return js, ctx, candidates, modules
|
||||
}
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
@@ -153,6 +154,30 @@ type delivered interface {
|
||||
Metadata() (*jetstream.MsgMetadata, error)
|
||||
}
|
||||
|
||||
// handleGroupEnd takes a group's end in (index.EndGroup): a body that does not read is terminated and said; an end
|
||||
// whose members are not all held yet, or while the build this program runs is not known, is left for redelivery.
|
||||
func handleGroupEnd(ctx context.Context, m delivered, ix *index.Indexer, h *index.Heard, logger *log.Logger) {
|
||||
end, err := delivery.DecodeForgeGroupEnd(m.Data())
|
||||
if err != nil {
|
||||
logger.Printf("%s: not a group's end, dropped: %v", m.Subject(), err)
|
||||
h.Outcome, h.Words = index.HeardDropped, err.Error()
|
||||
_ = m.Term()
|
||||
return
|
||||
}
|
||||
h.Kind, h.Branch = delivery.ForgeGroupEndEvent, end.Group
|
||||
applied, words, err := ix.EndGroup(ctx, end)
|
||||
h.Written, h.Late = applied.Written, applied.Late
|
||||
if err != nil {
|
||||
logger.Printf("%s: the end of the group %q left for redelivery: %v", m.Subject(), end.Group, err)
|
||||
h.Outcome, h.Words = index.HeardLeft, err.Error()
|
||||
return
|
||||
}
|
||||
h.Outcome, h.Words = index.HeardApplied, words
|
||||
if err := m.DoubleAck(ctx); err != nil {
|
||||
logger.Printf("%s: the end of the group %q applied, its acknowledgement was not confirmed: %v", m.Subject(), end.Group, err)
|
||||
}
|
||||
}
|
||||
|
||||
// handle takes one message of FORGE into the candidates and acknowledges it. A message it can never read is terminated,
|
||||
// never redelivered, and the log names its subject and why. A write that failed stays unacknowledged: the server
|
||||
// delivers it again (the server's ack wait is the only clock). Each message handled is written to the heard state
|
||||
@@ -171,6 +196,11 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto
|
||||
logger.Printf("%s: sequence %d is not written to the heard state: %v", h.Subject, h.Sequence, err)
|
||||
}
|
||||
}()
|
||||
// A delivery group's end (novox/hq ADR 0633 §4) has a body of its own: told apart by its name, before the decode.
|
||||
if strings.HasSuffix(m.Subject(), "."+delivery.ForgeGroupEndEvent) {
|
||||
handleGroupEnd(ctx, m, ix, &h, logger)
|
||||
return
|
||||
}
|
||||
f, err := delivery.DecodeForge(m.Data())
|
||||
if err != nil {
|
||||
logger.Printf("%s: not a forge message, dropped: %v", m.Subject(), err)
|
||||
@@ -182,7 +212,13 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto
|
||||
if f.Merge != "" {
|
||||
h.Commit = f.Merge
|
||||
}
|
||||
applied, err := ix.ApplyMessage(ctx, f)
|
||||
var applied index.Applied
|
||||
if f.Kind == delivery.ForgeMerged && h.Sequence > 0 {
|
||||
// A merge is taken in only by the indexer its revision names (ADR 0633 §1): kept in pending until then.
|
||||
applied, h.Words, err = ix.Arrive(ctx, h.Sequence, f)
|
||||
} else {
|
||||
applied, err = ix.ApplyMessage(ctx, f)
|
||||
}
|
||||
h.Written, h.Late = applied.Written, applied.Late
|
||||
switch {
|
||||
case errors.Is(err, index.ErrUnreadable):
|
||||
@@ -308,7 +344,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
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), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}}
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph},
|
||||
// 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{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)},
|
||||
SayTakeIn: sayTakeIn(js, cred)}
|
||||
// The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each
|
||||
// new one indexes the module it names when that module is registered and the bucket holds none (registered.go).
|
||||
var started atomic.Bool
|
||||
@@ -458,6 +498,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
defer pc.Stop()
|
||||
}
|
||||
// The pending revisions that name this build are taken in now, oldest first, before the consumer on the forge bridge's
|
||||
// stream starts (ADR 0633 §2); the build this program runs is asked until it is known, said every minute. A hand
|
||||
// take-in (indexer_take_in_pending) is answered meanwhile: the takeins watch runs already.
|
||||
if _, words, err := ix.TakeInAtStart(ctx, func(ctx context.Context) bool {
|
||||
// A context's bound between two asks of the node-engine, no timer (ADR 0437 §8), as askTakeIn bounds its wait.
|
||||
again, cancel := context.WithTimeout(ctx, time.Minute)
|
||||
defer cancel()
|
||||
<-again.Done()
|
||||
return ctx.Err() == nil
|
||||
}); err != nil {
|
||||
return fmt.Errorf("the take-in at start: %w", err)
|
||||
} else if words != "" {
|
||||
logger.Printf("take-in at start: still waiting: %s", words)
|
||||
}
|
||||
cons, err := cred.EventConsumer(ctx, js, forgeStream)
|
||||
if err != nil {
|
||||
// Essential: the forge's merges are what the indexer indexes.
|
||||
|
||||
@@ -102,6 +102,24 @@ func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 12
|
||||
if h := heardOf(t, heard); h.Outcome != index.HeardDropped || h.Sequence != 41 || h.Words == "" {
|
||||
t.Errorf("heard = %+v", h)
|
||||
}
|
||||
// 1585/T9 (ADR 0633 §4): a group's end is told apart by its name before the forge decode: one that reads is acked
|
||||
// (an indexer with no pending state takes the members in one by one), one that does not is dropped as a group's end.
|
||||
for _, c := range []struct {
|
||||
data string
|
||||
termd int
|
||||
logged string
|
||||
}{{`{"v":1,"group":"feat/x","members":[{"repository":"novox/mesh-sdk","number":3,"merge":"` + head + `"}]}`, 0, ""},
|
||||
{`{"v":1,"group":"feat/x"}`, 1, "not a group's end"}} {
|
||||
logs.Reset()
|
||||
m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event." + delivery.ForgeGroupEndEvent, data: []byte(c.data)}
|
||||
handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0))
|
||||
if m.termd != c.termd || m.acked+m.termd != 1 || !strings.Contains(logs.String(), c.logged) {
|
||||
t.Errorf("group end %s: terminated %d acked %d log %q", c.data, m.termd, m.acked, logs.String())
|
||||
}
|
||||
if h := heardOf(t, heard); h.Kind != delivery.ForgeGroupEndEvent && c.termd == 0 {
|
||||
t.Errorf("heard of a group's end = %+v", h)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value)
|
||||
|
||||
@@ -16,6 +16,8 @@ type takerIn interface {
|
||||
TakeIn(ctx context.Context, module string) index.TakenIn
|
||||
TakeInRepository(ctx context.Context, repository string) index.TakenIn
|
||||
JudgeAgain(ctx context.Context, module, digest string) index.TakenIn
|
||||
TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn
|
||||
DropPending(ctx context.Context, key, by, why string) index.TakenIn
|
||||
}
|
||||
|
||||
// takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the
|
||||
@@ -34,7 +36,13 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
return
|
||||
}
|
||||
asked, r := "take-in", index.TakenIn{}
|
||||
if a.Repository != "" {
|
||||
if a.Pending != "" && a.Drop {
|
||||
asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why)
|
||||
a.Module = a.Pending
|
||||
} else if a.Pending != "" {
|
||||
asked, r = "hand take-in of what waits", ix.TakeInPendingByHand(ctx, a.Pending, a.By, a.Why)
|
||||
a.Module = a.Pending
|
||||
} else if a.Repository != "" {
|
||||
asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository)
|
||||
a.Module = a.Repository
|
||||
} else if a.Judge {
|
||||
|
||||
@@ -62,3 +62,28 @@ func sayAccepted(js jetstream.JetStream, cred process.Account, logf func(string,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sayTakeIn publishes a take-in event on the indexer's own stream take-in (novox/hq ADR 0633 §5):
|
||||
// `mesh.stream.<module>.event.take-in.<kind>.<key>`. A publish that fails is the caller's to say.
|
||||
func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Context, delivery.TakeIn) error {
|
||||
return func(ctx context.Context, t delivery.TakeIn) error {
|
||||
event, err := delivery.TakeInEvent(t.Kind, t.Key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
data, err := delivery.EncodeTakeIn(t)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, event))
|
||||
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.Data = data
|
||||
pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = js.PublishMsg(pub, msg)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -121,6 +121,12 @@ type Indexer struct {
|
||||
ownMu sync.Mutex // guards ownFP
|
||||
ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild)
|
||||
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
|
||||
// Pending is the state of merges held for the indexer their revision names (pending.go, novox/hq ADR 0633 §1); nil
|
||||
// takes every merge in at once, as before the take-in step.
|
||||
Pending Store
|
||||
// SayTakeIn publishes a take-in event on the indexer's own stream take-in (ADR 0633 §5); nil says none.
|
||||
SayTakeIn func(ctx context.Context, t delivery.TakeIn) error
|
||||
pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go)
|
||||
}
|
||||
|
||||
func (ix *Indexer) logf(format string, args ...any) {
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"maps"
|
||||
"slices"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// overlayStore is a store whose writes stay in memory over the store it reads through (novox/hq ADR 0633 §1): a merge
|
||||
// applied to it is computed whole, its every fingerprint and verdict, and nothing reaches the bus. A key it wrote reads
|
||||
// as written, a key it deleted as absent, every other as the store under it holds it. Its compare-and-set holds against
|
||||
// what it reads: a key it wrote has a revision of its own, counted above every revision under it. A nil store under
|
||||
// it holds nothing.
|
||||
type overlayStore struct {
|
||||
under Store
|
||||
mu sync.Mutex
|
||||
vals map[string]overlaid
|
||||
next uint64
|
||||
}
|
||||
|
||||
type overlaid struct {
|
||||
value []byte
|
||||
revision uint64
|
||||
deleted bool
|
||||
}
|
||||
|
||||
func newOverlay(under Store) *overlayStore {
|
||||
return &overlayStore{under: under, vals: map[string]overlaid{}, next: 1 << 62}
|
||||
}
|
||||
|
||||
func (o *overlayStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) {
|
||||
o.mu.Lock()
|
||||
v, ok := o.vals[key]
|
||||
o.mu.Unlock()
|
||||
if ok {
|
||||
if v.deleted {
|
||||
return nil, v.revision, false, nil
|
||||
}
|
||||
return slices.Clone(v.value), v.revision, true, nil
|
||||
}
|
||||
if o.under == nil {
|
||||
return nil, 0, false, nil
|
||||
}
|
||||
return o.under.Get(ctx, key)
|
||||
}
|
||||
|
||||
// put checks the revision read against what the key holds now and writes it.
|
||||
func (o *overlayStore) put(ctx context.Context, key string, value []byte, revision uint64, create, del bool) error {
|
||||
_, rev, found, err := o.Get(ctx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
switch {
|
||||
case create && found, !create && (!found || rev != revision):
|
||||
return ErrConflict
|
||||
}
|
||||
o.next++
|
||||
o.vals[key] = overlaid{value: slices.Clone(value), revision: o.next, deleted: del}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (o *overlayStore) Create(ctx context.Context, key string, value []byte) error {
|
||||
return o.put(ctx, key, value, 0, true, false)
|
||||
}
|
||||
|
||||
func (o *overlayStore) Update(ctx context.Context, key string, value []byte, revision uint64) error {
|
||||
return o.put(ctx, key, value, revision, false, false)
|
||||
}
|
||||
|
||||
func (o *overlayStore) Delete(ctx context.Context, key string, revision uint64) error {
|
||||
return o.put(ctx, key, nil, revision, false, true)
|
||||
}
|
||||
|
||||
func (o *overlayStore) Keys(ctx context.Context) ([]string, error) {
|
||||
var keys []string
|
||||
if o.under != nil {
|
||||
var err error
|
||||
if keys, err = o.under.Keys(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
keys = slices.DeleteFunc(keys, func(k string) bool { _, ok := o.vals[k]; return ok })
|
||||
for k, v := range o.vals {
|
||||
if !v.deleted {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
slices.Sort(keys)
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
// written is what the overlay holds of key: its value, and whether it wrote or deleted it at all.
|
||||
func (o *overlayStore) written(key string) (value []byte, deleted, ok bool) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
v, ok := o.vals[key]
|
||||
return slices.Clone(v.value), v.deleted, ok
|
||||
}
|
||||
|
||||
// changed is every key the overlay wrote or deleted, in order.
|
||||
func (o *overlayStore) changed() []string {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
return slices.Sorted(maps.Keys(o.vals))
|
||||
}
|
||||
@@ -0,0 +1,861 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/graph"
|
||||
)
|
||||
|
||||
// The take-in step (novox/hq ADR 0633): a source revision is taken in only by the indexer it names. A merged message is
|
||||
// kept in the indexer's own state pending, one key per merge, until the take-in that ends it; a revision naming another
|
||||
// build of the indexer writes only what building and moving that indexer needs, and the rest stays pending for it.
|
||||
const (
|
||||
// PendingState is the indexer's state of held merges: the bucket mesh-delivery-indexer_pending, keyed
|
||||
// delivery.PendingKey, so the key order is the merge order. An old `state` entry (the indexer is foundation, ADR
|
||||
// 0595 §11), history 1, reached by the indexer alone.
|
||||
PendingState = "pending"
|
||||
PendingBucket = delivery.IndexerModule + "_" + PendingState
|
||||
// PendingTool reads it (ADR 0633 §5); TakeInPendingTool and DropPendingTool steer it.
|
||||
PendingTool = "indexer_pending"
|
||||
TakeInPendingTool = "indexer_take_in_pending"
|
||||
DropPendingTool = "indexer_drop_pending"
|
||||
)
|
||||
|
||||
// errClaimLost says another take-in claimed a pending merge first: this one took nothing in.
|
||||
var errClaimLost = errors.New("another take-in claimed the merges first, so this one took nothing in")
|
||||
|
||||
// NoBuild opens what a revision waits for when it gives the indexer no fingerprint: the reason follows. No indexer
|
||||
// runs such a build, so the revision waits until a later merge gives it one or a hand take-in (indexer_take_in_pending).
|
||||
const NoBuild = "no build: "
|
||||
|
||||
// ErrOwnBuildUnknown is what a merged message answers while the indexer cannot learn the build it runs (ADR 0633 §1):
|
||||
// the message is left unacknowledged, so the bus delivers it again, and nothing is written.
|
||||
var ErrOwnBuildUnknown = errors.New("the build of the indexer this program runs is not known")
|
||||
|
||||
// Pending is one value of the pending state: the merged message whole, as the forge bridge encoded it, the group it
|
||||
// was merged in (Ended once the group's end was heard), since when it is held, the indexer build its revision waits
|
||||
// for (absent while it waits for nothing yet, or behind a group being gathered), the build of the indexer that last
|
||||
// wrote it, the wait it was last said under on the take-in stream (Said, SaidAt), and Taking, set by a take-in that
|
||||
// claimed it: a second indexer that finds it claimed takes nothing in (ADR 0633 §1: never two take-ins of one merge).
|
||||
type Pending struct {
|
||||
V int `json:"v"`
|
||||
Forge json.RawMessage `json:"forge"`
|
||||
Group string `json:"group,omitempty"`
|
||||
Ended bool `json:"ended,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
WaitsFor string `json:"waits-for,omitempty"`
|
||||
LastWriter string `json:"last-writer"`
|
||||
Said string `json:"said,omitempty"`
|
||||
SaidKind string `json:"said-kind,omitempty"`
|
||||
SaidAt time.Time `json:"said-at,omitzero"`
|
||||
Taking string `json:"taking,omitempty"`
|
||||
}
|
||||
|
||||
// PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded.
|
||||
type PendingEntry struct {
|
||||
Key string
|
||||
Revision uint64
|
||||
Value Pending
|
||||
Forge delivery.Forge
|
||||
}
|
||||
|
||||
// DecodePending reads one pending value and its merged message.
|
||||
func DecodePending(data []byte) (Pending, delivery.Forge, error) {
|
||||
var p Pending
|
||||
if err := json.Unmarshal(data, &p); err != nil {
|
||||
return p, delivery.Forge{}, fmt.Errorf("not a pending value: %w", err)
|
||||
}
|
||||
if p.V != 1 {
|
||||
return p, delivery.Forge{}, fmt.Errorf("a pending value of version %d, not 1", p.V)
|
||||
}
|
||||
f, err := delivery.DecodeForge(p.Forge)
|
||||
if err != nil {
|
||||
return p, f, fmt.Errorf("its merged message: %w", err)
|
||||
}
|
||||
if f.Kind != delivery.ForgeMerged {
|
||||
return p, f, fmt.Errorf("a pending value holds a %s message, not a merged one", f.Kind)
|
||||
}
|
||||
return p, f, nil
|
||||
}
|
||||
|
||||
func encodePending(p Pending) ([]byte, error) {
|
||||
p.V = 1
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// ReadPending is every value the pending state holds, in key order (the merge order); a key that does not read is
|
||||
// said through logf and left.
|
||||
func ReadPending(ctx context.Context, st Store, logf func(string, ...any)) ([]PendingEntry, error) {
|
||||
keys, err := st.Keys(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listing %s: %w", PendingState, err)
|
||||
}
|
||||
slices.Sort(keys)
|
||||
var out []PendingEntry
|
||||
for _, key := range keys {
|
||||
value, rev, found, err := st.Get(ctx, key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s.%s: %w", PendingState, key, err)
|
||||
}
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
p, f, err := DecodePending(value)
|
||||
if err != nil {
|
||||
if logf != nil {
|
||||
logf("%s.%s is left: %v", PendingState, key, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
out = append(out, PendingEntry{Key: key, Revision: rev, Value: p, Forge: f})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Waits is the pending state read as what waits (ADR 0633 §3, §4): Ready are the merges no group still being gathered
|
||||
// holds back, in merge order, which are one revision together; Groups are the groups still being gathered, each with
|
||||
// its members held so far; Behind are the merges in a gathered group's member repositories that wait behind it, by the
|
||||
// group's take-in key.
|
||||
type Waits struct {
|
||||
Ready []PendingEntry
|
||||
Groups map[string][]PendingEntry
|
||||
Behind map[string][]PendingEntry
|
||||
}
|
||||
|
||||
// WaitsOf works out what waits from the entries, in key order. A member of a group whose end was not heard is
|
||||
// gathered; from the first such member on, a later merge in one of the group's members' repositories waits behind it,
|
||||
// because its tree holds that member's commit. Every other merge is ready.
|
||||
func WaitsOf(entries []PendingEntry) Waits {
|
||||
w := Waits{Groups: map[string][]PendingEntry{}, Behind: map[string][]PendingEntry{}}
|
||||
blocked := map[string]string{} // repository, lower case, to the group key it waits behind
|
||||
for _, e := range entries {
|
||||
if e.Value.Group != "" && !e.Value.Ended {
|
||||
w.Groups[e.Value.Group] = append(w.Groups[e.Value.Group], e)
|
||||
for _, m := range e.Forge.Members {
|
||||
if _, ok := blocked[strings.ToLower(m.Repository)]; !ok {
|
||||
blocked[strings.ToLower(m.Repository)] = delivery.GroupTakeInKey(e.Value.Group)
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
if g, ok := blocked[strings.ToLower(e.Forge.Repository)]; ok {
|
||||
w.Behind[g] = append(w.Behind[g], e)
|
||||
continue
|
||||
}
|
||||
w.Ready = append(w.Ready, e)
|
||||
}
|
||||
return w
|
||||
}
|
||||
|
||||
// mergesOf are the merges of entries, oldest first.
|
||||
func mergesOf(entries []PendingEntry) []delivery.GroupMerge {
|
||||
out := make([]delivery.GroupMerge, len(entries))
|
||||
for i, e := range entries {
|
||||
out[i] = delivery.GroupMerge{Repository: e.Forge.Repository, Number: e.Forge.Number, Merge: e.Forge.Merge}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func sinceOf(entries []PendingEntry) time.Time {
|
||||
var at time.Time
|
||||
for _, e := range entries {
|
||||
if at.IsZero() || e.Value.Since.Before(at) {
|
||||
at = e.Value.Since
|
||||
}
|
||||
}
|
||||
return at
|
||||
}
|
||||
|
||||
// groupWait is the take-in body of a group being gathered.
|
||||
func groupWait(kind, group string, held []PendingEntry) delivery.TakeIn {
|
||||
t := delivery.TakeIn{Kind: kind, Key: delivery.GroupTakeInKey(group), What: delivery.TakeInGroup, Group: group,
|
||||
Merges: mergesOf(held), Since: sinceOf(held)}
|
||||
if len(held) > 0 {
|
||||
t.Members = held[len(held)-1].Forge.Members
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// revisionWait is the take-in body of a revision: its key is its oldest merge's.
|
||||
func revisionWait(kind string, ready []PendingEntry, waitsFor string) delivery.TakeIn {
|
||||
return delivery.TakeIn{Kind: kind, Key: ready[0].Key, What: delivery.TakeInRevision, Merges: mergesOf(ready),
|
||||
WaitsFor: waitsFor, Since: sinceOf(ready)}
|
||||
}
|
||||
|
||||
// pendingOnBus says whether the pending state is on the bus: a state this delivery brings is bound when it appears
|
||||
// (issue 1790), and until then every merge is taken in at once, so the build this program runs is not needed. A read
|
||||
// that fails otherwise counts as on the bus: the step that follows answers the error.
|
||||
func (ix *Indexer) pendingOnBus(ctx context.Context) bool {
|
||||
_, _, _, err := ix.Pending.Get(ctx, delivery.PendingKey(0))
|
||||
return !errors.Is(err, process.ErrNotYet)
|
||||
}
|
||||
|
||||
// ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now.
|
||||
func (ix *Indexer) ownBuildNow(ctx context.Context) string {
|
||||
ix.ownMu.Lock()
|
||||
fp := ix.ownFP
|
||||
ix.ownMu.Unlock()
|
||||
if fp != "" || ix.OwnBuild == nil {
|
||||
return fp
|
||||
}
|
||||
ix.learnOwnBuild(ctx)
|
||||
ix.ownMu.Lock()
|
||||
defer ix.ownMu.Unlock()
|
||||
return ix.ownFP
|
||||
}
|
||||
|
||||
// sayTakeIn publishes a take-in event and keeps what it said on the entries, so indexer_pending reads the last event of
|
||||
// each wait from the state. A publish that fails is said in the log: the next evaluation says it again.
|
||||
func (ix *Indexer) sayTakeIn(ctx context.Context, t delivery.TakeIn) bool {
|
||||
if ix.SayTakeIn == nil {
|
||||
return true
|
||||
}
|
||||
if err := ix.SayTakeIn(ctx, t); err != nil {
|
||||
ix.logf("take-in: %s.%s is not said: %v", t.Kind, t.Key, err)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// markSaid writes on each entry the wait it was said under, by compare-and-set; a lost one is left (its next
|
||||
// evaluation says it again).
|
||||
func (ix *Indexer) markSaid(ctx context.Context, entries []PendingEntry, t *delivery.TakeIn, change func(*Pending)) {
|
||||
for _, e := range entries {
|
||||
p := e.Value
|
||||
if t != nil {
|
||||
p.Said, p.SaidKind, p.SaidAt = t.Key, t.Kind, ix.now()
|
||||
}
|
||||
if change != nil {
|
||||
change(&p)
|
||||
}
|
||||
data, err := encodePending(p)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if err := ix.Pending.Update(ctx, e.Key, data, e.Revision); err != nil {
|
||||
ix.logf("take-in: %s.%s is not written again: %v", PendingState, e.Key, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Arrive takes one merged message of the forge bridge's stream in under ADR 0633: seq is its stream sequence. It is
|
||||
// written to pending first (a redelivered message finds its key there and writes no second), and then what pending
|
||||
// holds is evaluated (Evaluate): taken in whole when its revision names the build this program runs, or held for the
|
||||
// indexer it names. It answers what was taken in, and a line saying what waits. A message is never written while the
|
||||
// build this program runs is not known (ErrOwnBuildUnknown).
|
||||
func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Applied, string, error) {
|
||||
if ix.Pending == nil || ix.Modules == nil || !ix.pendingOnBus(ctx) {
|
||||
// The state this delivery brings is not on the bus yet (issue 1790): the merge is taken in at once, as before,
|
||||
// whether or not the build this program runs is known.
|
||||
a, err := ix.ApplyMessage(ctx, f)
|
||||
return a, "", err
|
||||
}
|
||||
own := ix.ownBuildNow(ctx)
|
||||
if own == "" {
|
||||
return Applied{}, "", ErrOwnBuildUnknown
|
||||
}
|
||||
ix.pendingMu.Lock()
|
||||
defer ix.pendingMu.Unlock()
|
||||
raw, err := delivery.EncodeForge(f)
|
||||
if err != nil {
|
||||
return Applied{}, "", fmt.Errorf("%w: %v", ErrUnreadable, err)
|
||||
}
|
||||
key := delivery.PendingKey(seq)
|
||||
data, err := encodePending(Pending{Forge: raw, Group: f.Group, Since: ix.now(), LastWriter: own})
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
switch err := ix.Pending.Create(ctx, key, data); {
|
||||
case errors.Is(err, process.ErrNotYet):
|
||||
// The state this delivery brings is not on the bus yet (issue 1790): the merge is taken in at once, as before.
|
||||
ix.logf("take-in: the state %s is not on the bus yet, so %s#%d is taken in at once, as before ADR 0633", PendingBucket, f.Repository, f.Number)
|
||||
a, err := ix.ApplyMessage(ctx, f)
|
||||
return a, "", err
|
||||
case err != nil && !errors.Is(err, ErrConflict):
|
||||
return Applied{}, "", fmt.Errorf("%s.%s: %w", PendingState, key, err)
|
||||
}
|
||||
if f.Group != "" {
|
||||
entries, err := ReadPending(ctx, ix.Pending, ix.logf)
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
if held := WaitsOf(entries).Groups[f.Group]; len(held) > 0 {
|
||||
t := groupWait(delivery.TakeInPending, f.Group, held)
|
||||
if ix.sayTakeIn(ctx, t) {
|
||||
ix.markSaid(ctx, held, &t, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
return ix.evaluate(ctx, own, false)
|
||||
}
|
||||
|
||||
// EndGroup takes a group's end in (ADR 0633 §4): once every member it names is held, the group's members are marked
|
||||
// ended, the group's wait is said taken (take-in.taken under its group key), and pending is evaluated, the group now
|
||||
// one revision. A member named and neither held nor taken in yet answers an error, so the end is delivered again once
|
||||
// the member arrived. An end of a group taken in already (every member's merge at or behind its repository's taken
|
||||
// merge) changes nothing and is said: it is never delivered again for ever.
|
||||
func (ix *Indexer) EndGroup(ctx context.Context, end delivery.ForgeGroupEnd) (Applied, string, error) {
|
||||
if ix.Pending == nil || !ix.pendingOnBus(ctx) {
|
||||
return Applied{}, "no pending state: the group's merges were taken in one by one", nil
|
||||
}
|
||||
own := ix.ownBuildNow(ctx)
|
||||
if own == "" {
|
||||
return Applied{}, "", ErrOwnBuildUnknown
|
||||
}
|
||||
ix.pendingMu.Lock()
|
||||
defer ix.pendingMu.Unlock()
|
||||
entries, err := ReadPending(ctx, ix.Pending, ix.logf)
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
return Applied{}, "the pending state is not on the bus yet: the group's merges were taken in one by one", nil
|
||||
}
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
var held []PendingEntry
|
||||
for _, e := range entries {
|
||||
if e.Value.Group == end.Group {
|
||||
held = append(held, e)
|
||||
}
|
||||
}
|
||||
isHeld := func(m delivery.GroupMerge) bool {
|
||||
return slices.ContainsFunc(held, func(e PendingEntry) bool {
|
||||
return delivery.SameRepository(e.Forge.Repository, m.Repository) && e.Forge.Number == m.Number
|
||||
})
|
||||
}
|
||||
var missing []string
|
||||
for _, m := range end.Members {
|
||||
if isHeld(m) {
|
||||
continue
|
||||
}
|
||||
taken, err := ix.takenAlready(ctx, m)
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
if !taken {
|
||||
missing = append(missing, fmt.Sprintf("%s#%d", m.Repository, m.Number))
|
||||
}
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return Applied{}, "", fmt.Errorf("the end of the group %q names %s, not held yet: it waits for their merged messages", end.Group, strings.Join(missing, ", "))
|
||||
}
|
||||
if len(held) == 0 {
|
||||
ix.logf("take-in: the end of the group %q comes after its take-in: every member is taken in already, nothing changes", end.Group)
|
||||
return Applied{}, "the group is taken in already", nil
|
||||
}
|
||||
if !slices.ContainsFunc(held, func(e PendingEntry) bool { return !e.Value.Ended }) {
|
||||
return ix.evaluate(ctx, own, false)
|
||||
}
|
||||
// The group's wait ends first (take-in.taken under its group key); a revision it then is, pending for another
|
||||
// indexer, is said pending under its oldest merge's key by the evaluation that follows.
|
||||
t := groupWait(delivery.TakeInTaken, end.Group, held)
|
||||
said := &t
|
||||
if !ix.sayTakeIn(ctx, t) {
|
||||
said = nil // said again by the take-in or the hold that follows (saidWaits)
|
||||
}
|
||||
ix.markSaid(ctx, held, said, func(p *Pending) { p.Ended = true })
|
||||
ix.logf("take-in: the group %q ended with %d members held: one revision now", end.Group, len(held))
|
||||
return ix.evaluate(ctx, own, false)
|
||||
}
|
||||
|
||||
// takenAlready says whether a group member's merge is at or behind its repository's taken merge.
|
||||
func (ix *Indexer) takenAlready(ctx context.Context, m delivery.GroupMerge) (bool, error) {
|
||||
if ix.Taken == nil {
|
||||
return false, nil
|
||||
}
|
||||
key, err := delivery.RepositoryKey(m.Repository)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
value, _, found, err := ix.Taken.Get(ctx, key)
|
||||
if errors.Is(err, process.ErrNotYet) || (err == nil && !found) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("taken.%s: %w", key, err)
|
||||
}
|
||||
cur, err := delivery.DecodeTaken(value)
|
||||
if err != nil || cur.Merge == "" {
|
||||
return false, nil
|
||||
}
|
||||
if cur.Merge == m.Merge {
|
||||
return true, nil
|
||||
}
|
||||
return ix.older(ctx, m.Repository, m.Merge, cur.Merge)
|
||||
}
|
||||
|
||||
// TakeInAtStart is the take-in at the start (ADR 0633 §2), before the consumer on the forge bridge's stream starts:
|
||||
// it asks the build this program runs until it is known (wait says when to ask again, false ends it), then evaluates
|
||||
// pending, taking in the whole revisions that name this build; a group still being gathered stays, and a revision
|
||||
// naming yet another build is held for that one. A claim left by an indexer that ran before is void here.
|
||||
func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) bool) (Applied, string, error) {
|
||||
if ix.Pending == nil || !ix.pendingOnBus(ctx) {
|
||||
// Nothing can wait in a state not on the bus: the start asks no build and does not wait for one.
|
||||
return Applied{}, "the pending state is not on the bus yet: nothing waits", nil
|
||||
}
|
||||
own := ix.ownBuildNow(ctx)
|
||||
for own == "" {
|
||||
ix.logf("take-in: the build this program runs is not known yet, so nothing pending is taken in; asked again")
|
||||
if !wait(ctx) {
|
||||
return Applied{}, "", ErrOwnBuildUnknown
|
||||
}
|
||||
own = ix.ownBuildNow(ctx)
|
||||
}
|
||||
ix.pendingMu.Lock()
|
||||
defer ix.pendingMu.Unlock()
|
||||
a, words, err := ix.evaluate(ctx, own, true)
|
||||
if errors.Is(err, process.ErrNotYet) {
|
||||
return a, "the pending state is not on the bus yet: nothing waits", nil
|
||||
}
|
||||
return a, words, err
|
||||
}
|
||||
|
||||
// evaluate works out what pending's ready merges name, together (ADR 0633 §3), and takes them in or holds them.
|
||||
func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applied, string, error) {
|
||||
entries, err := ReadPending(ctx, ix.Pending, ix.logf)
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
w := WaitsOf(entries)
|
||||
if len(w.Ready) == 0 {
|
||||
return Applied{}, ix.waitingLine(w), nil
|
||||
}
|
||||
if !start {
|
||||
for _, e := range w.Ready {
|
||||
if e.Value.Taking != "" {
|
||||
ix.logf("take-in: %s.%s is claimed by the take-in %s, so this indexer takes nothing in", PendingState, e.Key, e.Value.Taking)
|
||||
return Applied{}, "claimed by another take-in", nil
|
||||
}
|
||||
}
|
||||
}
|
||||
names, dry, err := ix.computeOver(ctx, w.Ready)
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
if names == own {
|
||||
a, err := ix.takeInWhole(ctx, w.Ready, own, "")
|
||||
if errors.Is(err, errClaimLost) {
|
||||
return a, "claimed by another take-in", nil
|
||||
}
|
||||
return a, ix.waitingLine(w), err
|
||||
}
|
||||
return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, w.Ready, names, own, dry)
|
||||
}
|
||||
|
||||
// waitingLine says what waits after an evaluation, for the heard state.
|
||||
func (ix *Indexer) waitingLine(w Waits) string {
|
||||
var parts []string
|
||||
for _, g := range slices.Sorted(maps.Keys(w.Groups)) {
|
||||
parts = append(parts, fmt.Sprintf("the group %q gathered (%d held)", g, len(w.Groups[g])))
|
||||
}
|
||||
for _, g := range slices.Sorted(maps.Keys(w.Behind)) {
|
||||
parts = append(parts, fmt.Sprintf("%d merges behind %s", len(w.Behind[g]), g))
|
||||
}
|
||||
return strings.Join(parts, "; ")
|
||||
}
|
||||
|
||||
// dryRun is an indexer computing a take-in without writing it, and the overlays of what it would have written.
|
||||
type dryRun struct {
|
||||
ix *Indexer
|
||||
modules, graph, verdicts *overlayStore
|
||||
candidates, taken, predefined, configs *overlayStore
|
||||
oldSections *overlayStore
|
||||
}
|
||||
|
||||
// dry is an indexer over overlays of this one's states (overlay.go): what a take-in would write stays in memory and
|
||||
// nothing reaches the bus but the manifests and the packages the revision publishes, which clause 1 writes all the
|
||||
// same. It says no event, kicks nothing in the background, and knows what this one knows of the source, the toolchains
|
||||
// and the previews. Every field of Indexer is named here (TestDryNamesEveryField holds it so).
|
||||
func (ix *Indexer) dry() *dryRun {
|
||||
over := func(st Store) (Store, *overlayStore) {
|
||||
if st == nil {
|
||||
return nil, nil
|
||||
}
|
||||
o := newOverlay(st)
|
||||
return o, o
|
||||
}
|
||||
d := &dryRun{}
|
||||
x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum,
|
||||
GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains,
|
||||
ReadOpen: nil, OwnBuild: nil, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: 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)
|
||||
x.Graph, d.graph = over(ix.Graph)
|
||||
x.Verdicts, d.verdicts = over(ix.Verdicts)
|
||||
x.Taken, d.taken = over(ix.Taken)
|
||||
x.Predefined, d.predefined = over(ix.Predefined)
|
||||
x.ConfigCandidates, d.configs = over(ix.ConfigCandidates)
|
||||
x.OldSections, d.oldSections = over(ix.OldSections)
|
||||
ix.mu.Lock()
|
||||
x.source = sourceNow{rev: ix.source.rev, heads: maps.Clone(ix.source.heads)}
|
||||
x.previews = maps.Clone(ix.previews)
|
||||
x.toolchains = map[string]map[string]delivery.Toolchain{}
|
||||
for l, by := range ix.toolchains {
|
||||
x.toolchains[l] = maps.Clone(by)
|
||||
}
|
||||
x.moving = maps.Clone(ix.moving)
|
||||
x.untaken = maps.Clone(ix.untaken)
|
||||
ix.mu.Unlock()
|
||||
ix.ownMu.Lock()
|
||||
x.ownFP = ix.ownFP
|
||||
ix.ownMu.Unlock()
|
||||
d.ix = x
|
||||
return d
|
||||
}
|
||||
|
||||
// computeOver applies the ready merges, in merge order, to a dry indexer, as one overlay: each later merge's trees
|
||||
// govern (ADR 0633 §3). It answers the fingerprint the indexer's own module has once they are applied (none is "",
|
||||
// with its reason), and the dry run, whose overlays hold what the take-in would write. A merge that can never be read
|
||||
// is left out, as its take-in will drop it.
|
||||
func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (string, *dryRun, error) {
|
||||
d := ix.dry()
|
||||
for _, e := range ready {
|
||||
if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil {
|
||||
if errors.Is(err, ErrUnreadable) {
|
||||
continue
|
||||
}
|
||||
return "", d, fmt.Errorf("computing the take-in of %s (%s#%d): %w", e.Key, e.Forge.Repository, e.Forge.Number, err)
|
||||
}
|
||||
}
|
||||
key, _ := delivery.ModulesKey(delivery.IndexerModule)
|
||||
value, _, found, err := d.ix.Modules.Get(ctx, key)
|
||||
if err != nil {
|
||||
return "", d, err
|
||||
}
|
||||
if !found {
|
||||
return NoBuild + "the indexer holds no record of itself", d, nil
|
||||
}
|
||||
m, err := delivery.DecodeModule(value)
|
||||
if err != nil {
|
||||
return "", d, nil
|
||||
}
|
||||
if m.FP2 == "" {
|
||||
return NoBuild + m.None2, d, nil
|
||||
}
|
||||
return m.FP2, d, nil
|
||||
}
|
||||
|
||||
// claim marks the entries as taken in by this take-in, by compare-and-set on the revisions read; a lost one leaves
|
||||
// every claim made and answers false, and nothing is taken in.
|
||||
func (ix *Indexer) claim(ctx context.Context, entries []PendingEntry, own, by string) ([]PendingEntry, bool) {
|
||||
id := fmt.Sprintf("%s at %s", own, ix.now().Format(time.RFC3339Nano))
|
||||
if by != "" {
|
||||
id = by + ": " + id
|
||||
}
|
||||
claimed := make([]PendingEntry, 0, len(entries))
|
||||
for _, e := range entries {
|
||||
p := e.Value
|
||||
p.Taking, p.LastWriter = id, own
|
||||
data, err := encodePending(p)
|
||||
if err == nil {
|
||||
err = ix.Pending.Update(ctx, e.Key, data, e.Revision)
|
||||
}
|
||||
if err != nil {
|
||||
ix.logf("take-in: the claim of %s.%s is lost (%v): another take-in took it, so this one takes nothing in", PendingState, e.Key, err)
|
||||
return nil, false
|
||||
}
|
||||
_, rev, _, err := ix.Pending.Get(ctx, e.Key)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
e.Revision, e.Value = rev, p
|
||||
claimed = append(claimed, e)
|
||||
}
|
||||
return claimed, true
|
||||
}
|
||||
|
||||
// takeInWhole takes the entries in with this indexer, as merges are taken in (ApplyMessage), oldest first, then
|
||||
// removes them from pending and says each wait they were said under taken. by names a hand act ("" for none).
|
||||
func (ix *Indexer) takeInWhole(ctx context.Context, entries []PendingEntry, own, by string) (Applied, error) {
|
||||
return ix.takeInWholeBy(ctx, entries, own, by, "", "")
|
||||
}
|
||||
|
||||
// takeInWholeBy is takeInWhole; who and why, of a hand act, are carried on the take-in.taken events.
|
||||
func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, own, by, who, why string) (Applied, error) {
|
||||
claimed, ok := ix.claim(ctx, entries, own, by)
|
||||
if !ok {
|
||||
return Applied{}, errClaimLost
|
||||
}
|
||||
var applied Applied
|
||||
for _, e := range claimed {
|
||||
a, err := ix.ApplyMessage(ctx, e.Forge)
|
||||
applied.Written = append(applied.Written, a.Written...)
|
||||
if a.Late != "" {
|
||||
applied.Late = a.Late
|
||||
}
|
||||
if errors.Is(err, ErrUnreadable) {
|
||||
ix.logf("take-in: %s (%s#%d) can never be read, dropped: %v", e.Key, e.Forge.Repository, e.Forge.Number, err)
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return applied, fmt.Errorf("taking %s (%s#%d) in: %w", e.Key, e.Forge.Repository, e.Forge.Number, err)
|
||||
}
|
||||
}
|
||||
for _, e := range claimed {
|
||||
if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil {
|
||||
ix.logf("take-in: %s.%s is taken in and not removed: %v", PendingState, e.Key, err)
|
||||
}
|
||||
}
|
||||
ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key)
|
||||
for _, t := range saidWaits(claimed) {
|
||||
t.By, t.Why = who, why
|
||||
ix.sayTakeIn(ctx, t)
|
||||
}
|
||||
return applied, nil
|
||||
}
|
||||
|
||||
func byWords(by string) string {
|
||||
if by == "" {
|
||||
return ""
|
||||
}
|
||||
return " (a hand act: " + by + ")"
|
||||
}
|
||||
|
||||
// saidWaits are the taken events of every wait the entries were said under that is not ended yet.
|
||||
func saidWaits(entries []PendingEntry) []delivery.TakeIn {
|
||||
by := map[string][]PendingEntry{}
|
||||
for _, e := range entries {
|
||||
if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending {
|
||||
by[e.Value.Said] = append(by[e.Value.Said], e)
|
||||
}
|
||||
}
|
||||
var out []delivery.TakeIn
|
||||
for _, key := range slices.Sorted(maps.Keys(by)) {
|
||||
held := by[key]
|
||||
if strings.HasPrefix(key, "g.") {
|
||||
out = append(out, groupWait(delivery.TakeInTaken, held[0].Value.Group, held))
|
||||
continue
|
||||
}
|
||||
t := revisionWait(delivery.TakeInTaken, held, held[0].Value.WaitsFor)
|
||||
t.Key = key
|
||||
out = append(out, t)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// holdFor keeps the ready merges pending for the indexer build they name and writes only what building and moving
|
||||
// that indexer needs (ADR 0633 §1): the indexer's own module entry, its key of the module graph and the verdict on its
|
||||
// manifest, from the dry run; the manifests and the packages are written by the dry run itself. The revision is said
|
||||
// pending under its oldest merge's key, and a wait it was said under before by another key is said taken.
|
||||
func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own string, d *dryRun) error {
|
||||
key, _ := delivery.ModulesKey(delivery.IndexerModule)
|
||||
if err := copyKey(ctx, d.modules, ix.Modules, key); err != nil {
|
||||
return fmt.Errorf("the indexer's own module entry: %w", err)
|
||||
}
|
||||
if gk, err := graph.Key(delivery.IndexerModule); err == nil {
|
||||
if err := copyKey(ctx, d.graph, ix.Graph, gk); err != nil && !errors.Is(err, process.ErrNotYet) {
|
||||
return fmt.Errorf("the indexer's key of the module graph: %w", err)
|
||||
}
|
||||
}
|
||||
if value, _, found, err := d.ix.Modules.Get(ctx, key); err == nil && found {
|
||||
if m, err := delivery.DecodeModule(value); err == nil && m.Manifest != "" {
|
||||
if vk, err := delivery.VerdictsKey(m.Manifest); err == nil {
|
||||
if err := copyKey(ctx, d.verdicts, ix.Verdicts, vk); err != nil && !errors.Is(err, process.ErrNotYet) {
|
||||
return fmt.Errorf("the verdict on the indexer's manifest: %w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
t := revisionWait(delivery.TakeInPending, ready, names)
|
||||
var ended, unmarked []PendingEntry
|
||||
said, changed := false, false
|
||||
for _, e := range ready {
|
||||
switch {
|
||||
case e.Value.Said == t.Key && e.Value.SaidKind == delivery.TakeInPending:
|
||||
said = true
|
||||
changed = changed || e.Value.WaitsFor != names
|
||||
if e.Value.WaitsFor != names {
|
||||
unmarked = append(unmarked, e)
|
||||
}
|
||||
case e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending:
|
||||
ended = append(ended, e)
|
||||
unmarked = append(unmarked, e)
|
||||
default:
|
||||
unmarked = append(unmarked, e)
|
||||
}
|
||||
}
|
||||
for _, old := range saidWaits(ended) {
|
||||
ix.sayTakeIn(ctx, old)
|
||||
}
|
||||
if !said || changed {
|
||||
ix.logf("take-in: %d merges, oldest %s, name the indexer build %s, not %s, which runs: they wait for it", len(ready), t.Key, names, own)
|
||||
if !ix.sayTakeIn(ctx, t) {
|
||||
ix.markSaid(ctx, unmarked, nil, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own })
|
||||
return nil
|
||||
}
|
||||
unmarked = ready
|
||||
}
|
||||
ix.markSaid(ctx, unmarked, &t, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own })
|
||||
return nil
|
||||
}
|
||||
|
||||
// copyKey writes what the overlay holds of key to the store, when it wrote or deleted it.
|
||||
func copyKey(ctx context.Context, from *overlayStore, to Store, key string) error {
|
||||
if from == nil || to == nil {
|
||||
return nil
|
||||
}
|
||||
value, deleted, ok := from.written(key)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return write(ctx, to, key, func(cur []byte, found bool) ([]byte, bool, error) {
|
||||
switch {
|
||||
case deleted:
|
||||
return nil, found, nil
|
||||
case found && string(cur) == string(value):
|
||||
return nil, false, nil
|
||||
}
|
||||
return value, false, nil
|
||||
})
|
||||
}
|
||||
|
||||
// PendingWaits is what indexer_pending answers of the pending state's entries (ADR 0633 §5): the ready merges as one
|
||||
// revision, each group being gathered, and the merges behind a group, each with the modules it holds back and the
|
||||
// last take-in event said of it.
|
||||
func PendingWaits(entries []PendingEntry) []delivery.PendingWait {
|
||||
w := WaitsOf(entries)
|
||||
var out []delivery.PendingWait
|
||||
if len(w.Ready) > 0 {
|
||||
out = append(out, waitOf(w.Ready, delivery.TakeInRevision, ""))
|
||||
}
|
||||
for _, g := range slices.Sorted(maps.Keys(w.Groups)) {
|
||||
out = append(out, waitOf(w.Groups[g], delivery.TakeInGroup, ""))
|
||||
}
|
||||
for _, g := range slices.Sorted(maps.Keys(w.Behind)) {
|
||||
out = append(out, waitOf(w.Behind[g], delivery.TakeInRevision, g))
|
||||
}
|
||||
slices.SortStableFunc(out, func(a, b delivery.PendingWait) int { return a.Since.Compare(b.Since) })
|
||||
return out
|
||||
}
|
||||
|
||||
func waitOf(entries []PendingEntry, what, behind string) delivery.PendingWait {
|
||||
pw := delivery.PendingWait{Key: entries[0].Key, What: what, Merges: mergesOf(entries), Behind: behind, Since: sinceOf(entries),
|
||||
WaitsFor: entries[len(entries)-1].Value.WaitsFor, Holds: []string{}}
|
||||
if what == delivery.TakeInGroup {
|
||||
pw.Group = entries[0].Value.Group
|
||||
pw.Key = delivery.GroupTakeInKey(pw.Group)
|
||||
pw.Members = entries[len(entries)-1].Forge.Members
|
||||
pw.WaitsFor = ""
|
||||
}
|
||||
for _, e := range entries {
|
||||
pw.Keys = append(pw.Keys, e.Key)
|
||||
for _, raw := range e.Forge.Manifests {
|
||||
if name := manifestModule(raw); name != "" && !slices.Contains(pw.Holds, name) {
|
||||
pw.Holds = append(pw.Holds, name)
|
||||
}
|
||||
}
|
||||
if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) {
|
||||
pw.Said, pw.SaidAt = delivery.TakeInStream+"."+e.Value.SaidKind+"."+e.Value.Said, e.Value.SaidAt
|
||||
}
|
||||
}
|
||||
slices.Sort(pw.Holds)
|
||||
return pw
|
||||
}
|
||||
|
||||
// TakeInPendingByHand is the hand take-in (indexer_take_in_pending, ADR 0633 §5): the wait named by its key, the
|
||||
// ready revision (its m. key) or a group gathered so far (its g. key), is taken in with this indexer and its judge,
|
||||
// whatever build it names, for when the named indexer can never run or the group's end never comes. by and why are
|
||||
// said in the journal and on the take-in.taken event. What waited behind it is evaluated again after.
|
||||
func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string) TakenIn {
|
||||
r := TakenIn{Module: key}
|
||||
if ix.Pending == nil {
|
||||
r.Answer = "the pending state is not on the bus: nothing waits"
|
||||
return r
|
||||
}
|
||||
own := ix.ownBuildNow(ctx)
|
||||
ix.pendingMu.Lock()
|
||||
defer ix.pendingMu.Unlock()
|
||||
entries, err := ReadPending(ctx, ix.Pending, ix.logf)
|
||||
if err != nil {
|
||||
r.Answer = err.Error()
|
||||
return r
|
||||
}
|
||||
w := WaitsOf(entries)
|
||||
var take []PendingEntry
|
||||
switch {
|
||||
case strings.HasPrefix(key, "g."):
|
||||
for g, held := range w.Groups {
|
||||
if delivery.GroupTakeInKey(g) == key {
|
||||
take = held
|
||||
}
|
||||
}
|
||||
case len(w.Ready) > 0 && w.Ready[0].Key == key:
|
||||
take = w.Ready
|
||||
}
|
||||
if len(take) == 0 {
|
||||
r.Answer = fmt.Sprintf("nothing waits under %s: indexer_pending names the waits", key)
|
||||
return r
|
||||
}
|
||||
ix.logf("take-in: a hand take-in of %s (%d merges) asked by %s: %s", key, len(take), by, why)
|
||||
hand := fmt.Sprintf("%s (%s)", by, why)
|
||||
if _, err := ix.takeInWholeBy(ctx, take, own, hand, by, why); err != nil {
|
||||
r.Answer = err.Error()
|
||||
return r
|
||||
}
|
||||
r.Stored = true
|
||||
r.Answer = fmt.Sprintf("%d merges taken in by the running indexer %s", len(take), own)
|
||||
if _, _, err := ix.evaluate(ctx, own, false); err != nil {
|
||||
r.Answer += "; what waited behind it is not evaluated: " + err.Error()
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// DropPending is the hand drop (indexer_drop_pending, ADR 0633 §5): the one pending merge under key, abandoned
|
||||
// because a later merge replaced it, is removed and nothing of it is taken in. Its wait is said taken, naming the drop;
|
||||
// the merges left under that wait are said pending again by the evaluation that follows.
|
||||
func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn {
|
||||
r := TakenIn{Module: key}
|
||||
if ix.Pending == nil {
|
||||
r.Answer = "the pending state is not on the bus: nothing waits"
|
||||
return r
|
||||
}
|
||||
own := ix.ownBuildNow(ctx)
|
||||
ix.pendingMu.Lock()
|
||||
defer ix.pendingMu.Unlock()
|
||||
entries, err := ReadPending(ctx, ix.Pending, ix.logf)
|
||||
if err != nil {
|
||||
r.Answer = err.Error()
|
||||
return r
|
||||
}
|
||||
at := slices.IndexFunc(entries, func(e PendingEntry) bool { return e.Key == key })
|
||||
if at < 0 {
|
||||
r.Answer = fmt.Sprintf("%s holds no merge under %s", PendingState, key)
|
||||
return r
|
||||
}
|
||||
e := entries[at]
|
||||
if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil {
|
||||
r.Answer = fmt.Sprintf("%s.%s is not removed: %v", PendingState, key, err)
|
||||
return r
|
||||
}
|
||||
ix.logf("take-in: %s (%s#%d at %s) dropped by hand, asked by %s: %s; nothing of it is taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge, by, why)
|
||||
if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending {
|
||||
var same []PendingEntry
|
||||
for _, o := range entries {
|
||||
if o.Key != key && o.Value.Said == e.Value.Said {
|
||||
same = append(same, o)
|
||||
}
|
||||
}
|
||||
for _, t := range saidWaits([]PendingEntry{e}) {
|
||||
t.By, t.Why, t.Dropped = by, why, true
|
||||
ix.sayTakeIn(ctx, t)
|
||||
}
|
||||
ix.markSaid(ctx, same, nil, func(p *Pending) { p.Said, p.SaidKind = "", "" })
|
||||
}
|
||||
r.Stored, r.Answer = true, fmt.Sprintf("%s dropped: %s#%d at %s is not taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge)
|
||||
if own != "" {
|
||||
if _, _, err := ix.evaluate(ctx, own, false); err != nil {
|
||||
r.Answer += "; what waits is not evaluated again: " + err.Error()
|
||||
}
|
||||
}
|
||||
return r
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// takeInBed is an indexer holding the catalogue's avahi and the indexer's own module, both with a fingerprint, its
|
||||
// pending state and the take-in events it says; base is the catalogue merge it was built from.
|
||||
type takeInBed struct {
|
||||
ix *Indexer
|
||||
mods *memStore
|
||||
pending *memStore
|
||||
said []delivery.TakeIn
|
||||
base delivery.Forge
|
||||
logs *strings.Builder
|
||||
}
|
||||
|
||||
func newTakeInBed(t *testing.T) *takeInBed {
|
||||
t.Helper()
|
||||
ix, mods, f := newFormIndexerUnmerged(t, mod{"avahi", catalogueManifest(t, "avahi")},
|
||||
mod{delivery.IndexerModule, renamed(t, "avahi", delivery.IndexerModule)})
|
||||
if err := ix.Apply(context.Background(), f); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1")
|
||||
reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1")
|
||||
b := &takeInBed{ix: ix, mods: mods, pending: newMem(), base: f, logs: &strings.Builder{}}
|
||||
ix.Verdicts = newMem()
|
||||
ix.Pending = b.pending
|
||||
ix.SayTakeIn = func(_ context.Context, t delivery.TakeIn) error { b.said = append(b.said, t); return nil }
|
||||
logf := ix.Logf
|
||||
ix.Logf = func(format string, args ...any) {
|
||||
b.logs.WriteString(format + "\n")
|
||||
if logf != nil {
|
||||
logf(format, args...)
|
||||
}
|
||||
}
|
||||
own := b.fp(t, delivery.IndexerModule)
|
||||
if own == "" || b.fp(t, "avahi") == "" {
|
||||
t.Fatalf("the bed's modules have no fingerprint")
|
||||
}
|
||||
ix.OwnBuild = func(context.Context) (string, error) { return own, nil }
|
||||
return b
|
||||
}
|
||||
|
||||
func (b *takeInBed) fp(t *testing.T, module string) string {
|
||||
t.Helper()
|
||||
m, _ := moduleOf(t, b.mods, module)
|
||||
return m.FP2
|
||||
}
|
||||
|
||||
func (b *takeInBed) commit(t *testing.T, module string) string {
|
||||
t.Helper()
|
||||
m, _ := moduleOf(t, b.mods, module)
|
||||
return m.Commit
|
||||
}
|
||||
|
||||
// merge is a catalogue merge at commit c whose trees of avahi and the indexer are given ("" keeps the base's).
|
||||
func (b *takeInBed) merge(number int, c, avahi, indexer string) delivery.Forge {
|
||||
f := b.base
|
||||
f.Number, f.Merge, f.Head = number, c, c
|
||||
f.Trees = map[string]string{}
|
||||
for dir, tree := range b.base.Trees {
|
||||
f.Trees[dir] = tree
|
||||
}
|
||||
if avahi != "" {
|
||||
f.Trees["modules/avahi"] = avahi
|
||||
}
|
||||
if indexer != "" {
|
||||
f.Trees["modules/"+delivery.IndexerModule] = indexer
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
func (b *takeInBed) arrive(t *testing.T, seq uint64, f delivery.Forge) {
|
||||
t.Helper()
|
||||
if _, _, err := b.ix.Arrive(context.Background(), seq, f); err != nil {
|
||||
t.Fatalf("arrive %d: %v", seq, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (b *takeInBed) keys() []string {
|
||||
k, _ := b.pending.Keys(context.Background())
|
||||
slices.Sort(k)
|
||||
return k
|
||||
}
|
||||
|
||||
func (b *takeInBed) events() []string {
|
||||
var out []string
|
||||
for _, s := range b.said {
|
||||
out = append(out, s.Kind+" "+s.Key)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func sha(c byte) string { return strings.Repeat(string(c), 40) }
|
||||
|
||||
func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
ctx := context.Background()
|
||||
t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") }
|
||||
_, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", ""))
|
||||
if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') {
|
||||
t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi"))
|
||||
}
|
||||
// Before the pending state is on the bus nothing waits: the start asks no build, and the merge is taken in at once,
|
||||
// as before ADR 0633, though the build is unknown (TestIndexerWhole found the start waiting on it for ever).
|
||||
b.ix.Pending = &memStore{unbound: process.ErrNotYet}
|
||||
if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { t.Fatal("the start waits for the build"); return false }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")); err != nil || b.commit(t, "avahi") != sha('b') {
|
||||
t.Errorf("no pending state: err %v, avahi at %s: want the merge taken in at once", err, b.commit(t, "avahi"))
|
||||
}
|
||||
})
|
||||
t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
own, avahi0 := b.fp(t, delivery.IndexerModule), b.fp(t, "avahi")
|
||||
// A merge with no change to the indexer and nothing pending is taken in alone, as before.
|
||||
b.arrive(t, 1, b.merge(20, sha('b'), "tree-avahi-2", ""))
|
||||
if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 {
|
||||
t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events())
|
||||
}
|
||||
avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals)
|
||||
// T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending.
|
||||
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
|
||||
named := b.fp(t, delivery.IndexerModule)
|
||||
if named == own || b.fp(t, "avahi") != avahi1 || b.commit(t, "avahi") != sha('b') {
|
||||
t.Fatalf("indexer change: indexer %s (own %s), avahi %s at %s: want only the indexer written", named, own, b.fp(t, "avahi"), b.commit(t, "avahi"))
|
||||
}
|
||||
if g := graphOf(t, b.ix, delivery.IndexerModule); g.FP != named {
|
||||
t.Errorf("graph of the indexer names %s, want %s", g.FP, named)
|
||||
}
|
||||
if n := len(b.ix.Verdicts.(*memStore).vals); n > verdicts+1 {
|
||||
t.Errorf("verdicts %d, want at most %d: only the new indexer manifest's", n, verdicts+1)
|
||||
}
|
||||
if want := []string{delivery.PendingKey(2)}; !reflect.DeepEqual(b.keys(), want) || !slices.Equal(b.events(), []string{"pending " + delivery.PendingKey(2)}) || b.said[0].WaitsFor != named {
|
||||
t.Fatalf("pending %v, said %+v: want m.2 pending for %s", b.keys(), b.said, named)
|
||||
}
|
||||
// T5: the merge delivered again writes no second key and says nothing again.
|
||||
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
|
||||
if len(b.keys()) != 1 || len(b.said) != 1 {
|
||||
t.Errorf("redelivered: pending %v, said %v", b.keys(), b.events())
|
||||
}
|
||||
// A later merge joins: the newest governs and still names the new indexer.
|
||||
b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
|
||||
if len(b.keys()) != 2 || b.commit(t, "avahi") != sha('b') || len(b.said) != 1 {
|
||||
t.Fatalf("behind: pending %v, avahi at %s, said %v: want both pending and nothing said again", b.keys(), b.commit(t, "avahi"), b.events())
|
||||
}
|
||||
// A revert brings the indexer back to the running build: everything is taken in, oldest first.
|
||||
b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+delivery.IndexerModule]))
|
||||
if len(b.keys()) != 0 || b.fp(t, delivery.IndexerModule) != own || b.commit(t, "avahi") != sha('e') {
|
||||
t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, delivery.IndexerModule), b.commit(t, "avahi"))
|
||||
}
|
||||
if want := []string{"pending " + delivery.PendingKey(2), "taken " + delivery.PendingKey(2)}; !slices.Equal(b.events(), want) {
|
||||
t.Errorf("said %v, want %v", b.events(), want)
|
||||
}
|
||||
})
|
||||
t.Run("T10 the named indexer takes the revision in at its start; a gathered group stays", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
|
||||
g := b.merge(30, sha('f'), "", "")
|
||||
g.Repository, g.Trees, g.Manifests, g.Locks, g.Group = "novox/mesh-sdk", map[string]string{}, nil, nil, "feat/x"
|
||||
g.Members = []delivery.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}}
|
||||
b.arrive(t, 5, g)
|
||||
named := b.fp(t, delivery.IndexerModule)
|
||||
// The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits.
|
||||
next := &Indexer{}
|
||||
*next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
|
||||
next.toolchains = b.ix.toolchains
|
||||
asked := 0
|
||||
next.OwnBuild = func(context.Context) (string, error) {
|
||||
if asked++; asked < 2 {
|
||||
return "", errors.New("not yet")
|
||||
}
|
||||
return named, nil
|
||||
}
|
||||
if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") {
|
||||
t.Errorf("a hand ask while the own build is unknown: %+v", r)
|
||||
}
|
||||
if _, _, err := next.TakeInAtStart(ctx, func(context.Context) bool { return true }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if asked < 2 || !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') {
|
||||
t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi"))
|
||||
}
|
||||
})
|
||||
t.Run("T8 T9 T11 a group gathered until its end, a merge behind it, one alone, an end again", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
b.ix.Taken = newMem()
|
||||
// The commits in the order they were made: one is in another when it comes first.
|
||||
order := "a756bc"
|
||||
b.ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) {
|
||||
return strings.IndexByte(order, commit[0]) <= strings.IndexByte(order, in[0]), nil
|
||||
}
|
||||
member := func(repo string, number int, c string) delivery.Forge {
|
||||
f := b.merge(number, c, "", "")
|
||||
f.Group = "feat/take-in"
|
||||
f.Members = []delivery.Head{{Repository: catalogueRepo, Number: 40, Head: sha('1')}, {Repository: "novox/mesh-sdk", Number: 41, Head: sha('2')}}
|
||||
if repo != catalogueRepo {
|
||||
f.Repository, f.Trees, f.Manifests, f.Locks = repo, map[string]string{}, nil, nil
|
||||
}
|
||||
f.Head = map[int]string{40: sha('1'), 41: sha('2')}[number]
|
||||
return f
|
||||
}
|
||||
b.arrive(t, 1, member(catalogueRepo, 40, sha('7')))
|
||||
if b.commit(t, "avahi") == sha('7') || !slices.Equal(b.events(), []string{"pending g.feat/take-in"}) {
|
||||
t.Fatalf("first member: avahi at %s, said %v: want held, the group said pending", b.commit(t, "avahi"), b.events())
|
||||
}
|
||||
// A merge in another repository is taken in alone; one in a member's repository waits behind the group.
|
||||
alone := b.merge(50, sha('5'), "", "")
|
||||
alone.Repository, alone.Trees, alone.Manifests, alone.Locks = "novox/hq", map[string]string{}, nil, nil
|
||||
b.arrive(t, 2, alone)
|
||||
b.arrive(t, 3, b.merge(51, sha('6'), "tree-avahi-9", ""))
|
||||
if !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(1), delivery.PendingKey(3)}) {
|
||||
t.Fatalf("pending %v: want the member and the merge behind it", b.keys())
|
||||
}
|
||||
end := delivery.ForgeGroupEnd{Group: "feat/take-in", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('7')}, {Repository: "novox/mesh-sdk", Number: 41, Merge: sha('b')}}}
|
||||
if _, _, err := b.ix.EndGroup(ctx, end); err == nil {
|
||||
t.Fatal("an end before its last member is taken in")
|
||||
}
|
||||
b.arrive(t, 4, member("novox/mesh-sdk", 41, sha('b')))
|
||||
if _, _, err := b.ix.EndGroup(ctx, end); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') {
|
||||
t.Fatalf("end: pending %v, avahi at %s: want the group and the merge behind it taken in", b.keys(), b.commit(t, "avahi"))
|
||||
}
|
||||
if want := []string{"pending g.feat/take-in", "pending g.feat/take-in", "taken g.feat/take-in"}; !slices.Equal(b.events(), want) {
|
||||
t.Errorf("said %v, want %v", b.events(), want)
|
||||
}
|
||||
// R1: the end published again after the take-in changes nothing and is not delivered again.
|
||||
if _, words, err := b.ix.EndGroup(ctx, end); err != nil || !strings.Contains(words, "taken in already") {
|
||||
t.Errorf("an end again: %q %v", words, err)
|
||||
}
|
||||
// T11 (the lead's row): a group whose revision changes the indexer: the group's wait is taken, then the revision pending.
|
||||
b.said = nil
|
||||
g := member(catalogueRepo, 40, sha('c'))
|
||||
g.Group, g.Trees = "feat/two", b.merge(0, "", "", "tree-indexer-7").Trees
|
||||
g.Members = g.Members[:1]
|
||||
b.arrive(t, 9, g)
|
||||
if _, _, err := b.ix.EndGroup(ctx, delivery.ForgeGroupEnd{Group: "feat/two", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('c')}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := []string{"pending g.feat/two", "taken g.feat/two", "pending " + delivery.PendingKey(9)}; !slices.Equal(b.events(), want) {
|
||||
t.Errorf("a group naming another indexer: said %v, want %v", b.events(), want)
|
||||
}
|
||||
})
|
||||
t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
|
||||
named := b.fp(t, delivery.IndexerModule)
|
||||
b.ix.ownMu.Lock()
|
||||
b.ix.ownFP = named
|
||||
b.ix.ownMu.Unlock()
|
||||
// The other indexer claims the merge between this one's read and its claim.
|
||||
b.pending.beforeWrite = func() {
|
||||
v := b.pending.vals[delivery.PendingKey(2)]
|
||||
var p Pending
|
||||
_ = json.Unmarshal(v, &p)
|
||||
p.Taking = "the other indexer"
|
||||
data, _ := json.Marshal(p)
|
||||
b.pending.put(delivery.PendingKey(2), data)
|
||||
}
|
||||
if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { return false }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if b.commit(t, "avahi") == sha('c') || len(b.keys()) != 1 || !strings.Contains(b.logs.String(), "is lost") {
|
||||
t.Errorf("avahi at %s, pending %v: want nothing taken in by the loser, and it said", b.commit(t, "avahi"), b.keys())
|
||||
}
|
||||
})
|
||||
t.Run("T12 T13 indexer_pending names the wait; a hand take-in and a drop say what they ended", func(t *testing.T) {
|
||||
b := newTakeInBed(t)
|
||||
b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
|
||||
b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
|
||||
var answer delivery.PendingAnswer
|
||||
for _, tool := range PendingTools(b.pending, nil, nil) {
|
||||
if tool.Name == PendingTool {
|
||||
out, err := tool.Run(map[string]any{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
answer = out.(delivery.PendingAnswer)
|
||||
}
|
||||
}
|
||||
if len(answer.Waits) != 1 || answer.Waits[0].Key != delivery.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", delivery.IndexerModule}) ||
|
||||
answer.Waits[0].Said != "take-in.pending."+delivery.PendingKey(2) || len(answer.Waits[0].Keys) != 2 {
|
||||
t.Fatalf("indexer_pending = %+v", answer)
|
||||
}
|
||||
b.said = nil
|
||||
if r := b.ix.DropPending(ctx, delivery.PendingKey(2), "lead@shanks", "replaced by m.3"); !r.Stored {
|
||||
t.Fatalf("drop: %+v", r)
|
||||
}
|
||||
if len(b.said) < 2 || !b.said[0].Dropped || b.said[0].Key != delivery.PendingKey(2) || b.said[0].By != "lead@shanks" || b.said[1].Key != delivery.PendingKey(3) {
|
||||
t.Fatalf("drop said %+v: want m.2 taken (dropped, by) and m.3 pending", b.said)
|
||||
}
|
||||
b.said = nil
|
||||
if r := b.ix.TakeInPendingByHand(ctx, delivery.PendingKey(3), "lead@shanks", "the named indexer can never run"); !r.Stored {
|
||||
t.Fatalf("hand take-in: %+v", r)
|
||||
}
|
||||
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(b.said) != 1 || b.said[0].Kind != delivery.TakeInTaken || b.said[0].By != "lead@shanks" {
|
||||
t.Errorf("hand take-in: pending %v, avahi at %s, said %+v", b.keys(), b.commit(t, "avahi"), b.said)
|
||||
}
|
||||
})
|
||||
t.Run("the dry run names every field of the indexer", func(t *testing.T) {
|
||||
// dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails.
|
||||
named := map[string]bool{}
|
||||
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections",
|
||||
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
|
||||
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Pending", "SayTakeIn"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
for i := range ty.NumField() {
|
||||
if f := ty.Field(i); f.IsExported() && !named[f.Name] {
|
||||
t.Errorf("Indexer.%s is not named by dry", f.Name)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// PendingTools are the indexer's tools of what waits for a take-in (novox/hq ADR 0633 §5): indexer_pending reads the
|
||||
// pending state, and indexer_take_in_pending and indexer_drop_pending steer it by an ask in takeins, as indexer_take_in
|
||||
// does. pending is the pending state as the tools bundle reads it; ask writes an ask and waits for the indexer's answer.
|
||||
func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool {
|
||||
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
|
||||
hand := func(drop bool) func(map[string]any) (any, error) {
|
||||
return func(args map[string]any) (any, error) {
|
||||
key, value, err := NewPendingAsk(str(args, "key"), str(args, "why"), stdio.CallerOf(args), drop, now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a, err := ask(key, value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Result, nil
|
||||
}
|
||||
}
|
||||
return []stdio.Tool{
|
||||
{Name: PendingTool,
|
||||
Description: "What waits for a take-in (ADR 0633): merges the indexer holds in its pending state instead of taking them in. " +
|
||||
"A revision (what revision, its key its oldest merge's m.<sequence>) waits for the build of the indexer it names " +
|
||||
"(waits-for): the running indexer wrote only what building and moving that indexer needs, and the new indexer takes " +
|
||||
"it in at its start. A group (what group, key g.<group>) is a delivery group whose members are gathered until the " +
|
||||
"forge bridge says its end; a merge in one of its members' repositories waits behind it (behind). Each names its " +
|
||||
"merges, since when it waits, the modules it holds back (holds) and the last take-in event said of it. Empty waits: " +
|
||||
"every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " +
|
||||
"indexer_take_in_pending and indexer_drop_pending steer a wait.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{}},
|
||||
Run: func(map[string]any) (any, error) {
|
||||
var unread []string
|
||||
entries, err := ReadPending(context.Background(), pending, func(format string, args ...any) {
|
||||
unread = append(unread, fmt.Sprintf(format, args...))
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return delivery.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}, nil
|
||||
}},
|
||||
{Name: TakeInPendingTool,
|
||||
Description: "Take a wait in by hand with the running indexer and its judge (ADR 0633 §5): the revision under its m. key, " +
|
||||
"whatever build of the indexer it names, or a delivery group under its g. key with the members gathered so far. For " +
|
||||
"when the named indexer can never run, or a group's end never comes. A hand act: who and why are said in the " +
|
||||
"indexer's journal and on the take-in.taken event, which clears take-in-waits. indexer_pending names the keys.",
|
||||
Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{
|
||||
"key": map[string]any{"type": "string", "description": "the wait's key, m.<sequence> or g.<group>, as indexer_pending names it"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is taken in by hand, in plain words"}}},
|
||||
Run: hand(false)},
|
||||
{Name: DropPendingTool,
|
||||
Description: "Drop one pending merge by hand, abandoned because a later merge replaced it (ADR 0633 §5): it is removed from " +
|
||||
"the pending state and nothing of it is taken in; what waited with it is worked out again. A hand act: who and why are " +
|
||||
"said in the journal and on the take-in.taken event (dropped).",
|
||||
Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{
|
||||
"key": map[string]any{"type": "string", "description": "the pending merge's key, m.<sequence>, as indexer_pending names its merges' wait"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
|
||||
Run: hand(true)},
|
||||
}
|
||||
}
|
||||
@@ -51,6 +51,17 @@ func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing.
|
||||
}
|
||||
|
||||
func TestAMergeHeardAfterTheSourceNamesItsHeadIsStampedWhenWritten(t *testing.T) { // 1344/AC1 (the source first)
|
||||
// A source revision older than the one held is left (merged here from a test of its own, for the test budget).
|
||||
t.Run("an older source revision is left", func(t *testing.T) {
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
ctx := context.Background()
|
||||
_ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
|
||||
_ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge))
|
||||
_ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge))
|
||||
if m, _ := moduleOf(t, mods, "a"); m.Src != 9 {
|
||||
t.Errorf("module a Src = %d, want 9", m.Src)
|
||||
}
|
||||
})
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
ctx := context.Background()
|
||||
if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil {
|
||||
@@ -93,14 +104,3 @@ func TestAMergeThatNoSourceRevisionNamesIsWrittenAsToday(t *testing.T) { // 1344
|
||||
t.Errorf("module a = %+v, want the new merge with Src 0", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestASourceRevisionOlderThanTheOneHeldIsLeft(t *testing.T) {
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
ctx := context.Background()
|
||||
_ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
|
||||
_ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge))
|
||||
_ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge))
|
||||
if m, _ := moduleOf(t, mods, "a"); m.Src != 9 {
|
||||
t.Errorf("module a Src = %d, want 9", m.Src)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,7 +33,11 @@ type TakeInAsk struct {
|
||||
// Judge asks the manifests to be judged again now (indexer_judge_again) instead of the module to be taken in.
|
||||
Judge bool `json:"judge,omitempty"`
|
||||
// Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest).
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
// Pending asks the wait under this take-in key to be taken in by hand (indexer_take_in_pending), or with Drop the
|
||||
// one pending merge under it to be dropped (indexer_drop_pending), instead of a module (pending.go, ADR 0633 §5).
|
||||
Pending string `json:"pending,omitempty"`
|
||||
Drop bool `json:"drop,omitempty"`
|
||||
By string `json:"by,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
Result *TakenIn `json:"result,omitempty"`
|
||||
@@ -52,8 +56,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) {
|
||||
if a.V != 1 {
|
||||
return a, fmt.Errorf("a takeins value of version %d, not 1", a.V)
|
||||
}
|
||||
if a.Module == "" && a.Repository == "" {
|
||||
return a, errors.New("a takeins value naming no module or repository")
|
||||
if a.Module == "" && a.Repository == "" && a.Pending == "" {
|
||||
return a, errors.New("a takeins value naming no module, repository or pending wait")
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
@@ -82,3 +86,25 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by
|
||||
data, err := json.Marshal(TakeInAsk{V: 1, Repository: repository, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
|
||||
return "repository." + key, data, err
|
||||
}
|
||||
|
||||
// NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key
|
||||
// in takeins, `pending.<take-in|drop>.<key>`, and the value. key is a wait's key as indexer_pending names it, for a
|
||||
// drop one pending merge's.
|
||||
func NewPendingAsk(key, why, by string, drop bool, at time.Time) (string, []byte, error) {
|
||||
key = strings.TrimSpace(key)
|
||||
if !strings.HasPrefix(key, "m.") && !(strings.HasPrefix(key, "g.") && !drop) {
|
||||
if drop {
|
||||
return "", nil, errors.New("name the pending merge to drop by its key, m.<sequence>, as indexer_pending names it")
|
||||
}
|
||||
return "", nil, errors.New("name the wait to take in by its key, m.<sequence> or g.<group>, as indexer_pending names it")
|
||||
}
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return "", nil, errors.New("say why (why): a hand act on what waits is said in the journal and on its take-in event")
|
||||
}
|
||||
act := "take-in"
|
||||
if drop {
|
||||
act = "drop"
|
||||
}
|
||||
data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, Drop: drop, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
|
||||
return "pending." + act + "." + key, data, err
|
||||
}
|
||||
|
||||
@@ -16,6 +16,8 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/graph"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// The controller's manifest reader is mesh-controller internal/catalogue (Manifest, manifest.go: Consumes :521,
|
||||
@@ -28,6 +30,12 @@ type state struct {
|
||||
History int `json:"history,omitempty"`
|
||||
}
|
||||
|
||||
type streamDecl struct {
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
MaxAgeDays int `json:"max-age-days,omitempty"`
|
||||
}
|
||||
|
||||
type artifact struct {
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
@@ -51,6 +59,7 @@ type manifest struct {
|
||||
Emits []string `json:"emits"`
|
||||
Consumes []string `json:"consumes"`
|
||||
State []state `json:"state"`
|
||||
Streams []streamDecl `json:"streams"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
@@ -99,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
}
|
||||
// 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: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory},
|
||||
{Name: graph.State, History: graph.History}}; !reflect.DeepEqual(m.State, want) {
|
||||
{Name: graph.State, History: graph.History}, {Name: index.PendingState, 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-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
@@ -125,7 +134,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
|
||||
// reads the verdicts state (indexer_verdicts) and asks the indexer to judge again through the takeins state
|
||||
// (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections).
|
||||
if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge"}; !reflect.DeepEqual(m.Tools, want) {
|
||||
// 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending
|
||||
// ask through takeins.
|
||||
if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool}; !reflect.DeepEqual(m.Tools, want) {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||
}
|
||||
// ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a
|
||||
@@ -134,8 +145,13 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) {
|
||||
t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims)
|
||||
}
|
||||
// No streams, no bus, no other seat (ADR 0437 §4); it says no event.
|
||||
for _, k := range []string{"streams", "logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
|
||||
// 1585/T11 (ADR 0633 §1, §5): its one stream, take-in, an old streams entry (the indexer is foundation, ADR 0595 §11),
|
||||
// carries the take-in events.
|
||||
if want := []streamDecl{{Name: delivery.TakeInStream, Events: []string{"take-in.**"}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) {
|
||||
t.Errorf("streams = %+v, want %+v", m.Streams, want)
|
||||
}
|
||||
// No other bus section, no other seat (ADR 0437 §4).
|
||||
for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the indexer has none", k)
|
||||
}
|
||||
|
||||
@@ -18,7 +18,10 @@
|
||||
"indexer_take_in",
|
||||
"indexer_old_sections",
|
||||
"indexer_verdicts",
|
||||
"indexer_judge"
|
||||
"indexer_judge",
|
||||
"indexer_pending",
|
||||
"indexer_take_in_pending",
|
||||
"indexer_drop_pending"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-build-go.toolchain.reported",
|
||||
@@ -38,7 +41,11 @@
|
||||
{ "name": "verdicts", "history": 5 },
|
||||
{ "name": "oldsections", "history": 1 },
|
||||
{ "name": "packages", "history": 5 },
|
||||
{ "name": "graph", "history": 5 }
|
||||
{ "name": "graph", "history": 5 },
|
||||
{ "name": "pending", "history": 1 }
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "take-in", "events": ["take-in.**"], "max-age-days": 7 }
|
||||
],
|
||||
"emits": [
|
||||
"manifest-refused",
|
||||
|
||||
Reference in New Issue
Block a user