IX24: the indexer's four kept calls are asked again only while their ask is open
messageVia, stepsVia, sourceVia and versionsVia go through CallKeptWhileOpen: open while the commit's take-in is pending, the candidate is held, the branch is watched, the module is in the newest view. A call that ends with its ask closed is index.ErrAskClosed, not an unanswered call. Builds: design 47 G §351, §367, §460, §531; A-190 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -14,14 +14,16 @@ import (
|
||||
const mergedTool = index.ForgeBridgeModule + ".bridge_merged_message"
|
||||
|
||||
// messageVia asks the forge bridge for a merge's merged message, for a held merge the indexer does not hold in memory (a
|
||||
// restart, a new build). An error is the bridge's own words.
|
||||
func messageVia(call toolCaller) func(ctx context.Context, repository, merge string, number int) ([]byte, error) {
|
||||
return func(_ context.Context, repository, merge string, number int) ([]byte, error) {
|
||||
// restart, a new build), a kept call asked again only while the commit's take-in is pending (A-190). An error is the
|
||||
// bridge's own words.
|
||||
func messageVia(call openCaller, pending func(ctx context.Context, repository, merge string) (bool, error), logf func(string, ...any)) func(ctx context.Context, repository, merge string, number int) ([]byte, error) {
|
||||
return func(ctx context.Context, repository, merge string, number int) ([]byte, error) {
|
||||
ask := map[string]any{"repository": repository, "merge": merge}
|
||||
if number > 0 {
|
||||
ask["number"] = number
|
||||
}
|
||||
raw, err := call(mergedTool, ask)
|
||||
raw, err := keptWhileOpen(ctx, call, logf, "message."+repository+"."+merge, mergedTool, ask,
|
||||
func(ctx context.Context) (bool, error) { return pending(ctx, repository, merge) })
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// openCaller is process.Conn.CallKeptWhileOpen: a kept call asked again only while its ask is open (design 47 G §351,
|
||||
// §460, §531; A-190).
|
||||
type openCaller func(o busretry.Open, key string, body any) (json.RawMessage, error)
|
||||
|
||||
// keptWhileOpen asks verb as a kept call keyed by key, asked again only while open says what the ask serves still stands.
|
||||
// A call that ends with its ask closed is index.ErrAskClosed, never an unanswered call (A-190).
|
||||
func keptWhileOpen(ctx context.Context, call openCaller, logf func(string, ...any), key, verb string, body any,
|
||||
open func(context.Context) (bool, error)) (json.RawMessage, error) {
|
||||
raw, err := call(busretry.Open{Key: key, Open: func(string) (bool, error) { return open(ctx) }, Logf: logf}, verb, body)
|
||||
if err == nil {
|
||||
return raw, nil
|
||||
}
|
||||
if still, oerr := open(ctx); oerr == nil && !still {
|
||||
return nil, fmt.Errorf("%s: %w", verb, index.ErrAskClosed)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// A-190: the four asks pass their key's open function to the kept call; a call that fails with its key closed is
|
||||
// index.ErrAskClosed, one that fails while it is open keeps its own error.
|
||||
func TestKeptWhileOpenNamesAClosedAsk(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
failing := errors.New("no responders")
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
open bool
|
||||
answer error
|
||||
want error
|
||||
}{{"open", true, failing, failing}, {"closed", false, failing, index.ErrAskClosed}, {"answered", false, nil, nil}} {
|
||||
var asked busretry.Open
|
||||
call := func(o busretry.Open, key string, _ any) (json.RawMessage, error) {
|
||||
asked = o
|
||||
return json.RawMessage(`{}`), c.answer
|
||||
}
|
||||
_, err := keptWhileOpen(ctx, call, t.Logf, "k1", "tool", nil, func(context.Context) (bool, error) { return c.open, nil })
|
||||
if !errors.Is(err, c.want) {
|
||||
t.Errorf("%s: err = %v, want %v", c.name, err, c.want)
|
||||
}
|
||||
if got, _ := asked.Open("k1"); asked.Key != "k1" || got != c.open {
|
||||
t.Errorf("%s: the kept call was given key %q open %v", c.name, asked.Key, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -449,7 +449,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph,
|
||||
SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
|
||||
Message: messageVia(calls.CallKept), Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
|
||||
Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
|
||||
// Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go).
|
||||
SayCatalogRevision: afterRevision(refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow), &fillFollow),
|
||||
SayFill: sayFill(conn, cred),
|
||||
@@ -469,11 +469,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.RegistrationsBucket, "no registration is kept or written", logger.Printf)},
|
||||
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)},
|
||||
routerRecordOf(cred.Module, calls.Call, calls.Ask), ix, logger.Printf)
|
||||
ix.Message = messageVia(calls.CallKeptWhileOpen, ix.TakeInPending, logger.Printf) // a kept call, called again through a transport fault while the commit's take-in is pending (ADR 0668 §1, A-190)
|
||||
registry.Pruned = ix.PruneBranchBuilds
|
||||
ix.Registry = registry
|
||||
named.Changed = ix.RemoveUnlisted
|
||||
// The first fill of the versions (design 47 G §372): at the start, and again at each catalog revision kept until it is made.
|
||||
fill := &versionsFill{ix: ix, ctx: ctx, ask: versionsVia(calls.CallKept), logf: logger.Printf}
|
||||
fill := &versionsFill{ix: ix, ctx: ctx, ask: versionsVia(calls.CallKeptWhileOpen, ix.ModuleInNewestView, logger.Printf), logf: logger.Printf}
|
||||
fillFollow.Store(fill)
|
||||
fill.kick()
|
||||
ix.SeedModules = func(ctx context.Context) ([]index.SeedModule, error) {
|
||||
@@ -481,7 +482,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if r == nil {
|
||||
return nil, errors.New("the seat list is not read yet: the seed waits for it")
|
||||
}
|
||||
return ix.SeedModulesFrom(ctx, r.Definitions(), stepsVia(calls.CallKept))
|
||||
return ix.SeedModulesFrom(ctx, r.Definitions(), stepsVia(calls.CallKeptWhileOpen, ix.SeedEntryHeld, logger.Printf))
|
||||
}
|
||||
// Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the
|
||||
// node-engine names the successor while it replaces this process, and an old process that learnt it later would take
|
||||
@@ -518,7 +519,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// The source revision of a watched branch (design 47 G §15, §115, §362): asked of the seat mesh-delivery's verb source as a
|
||||
// kept call, once at the start, before the handler of its event source-moved reads anything; a source-moved heard
|
||||
// meanwhile is read after.
|
||||
source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, applyBranch: ix.ApplyBranchSource, logf: logger.Printf}
|
||||
source := &sourceFollow{read: sourceVia(ctx, calls.CallKeptWhileOpen, ix.BranchWatched, logger.Printf), apply: ix.ApplySource, applyBranch: ix.ApplyBranchSource, logf: logger.Printf}
|
||||
go source.catchUp(ctx)
|
||||
// Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the
|
||||
// assignments is settled first, so the toolchains settle through the gate already knowing it.
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"log"
|
||||
"reflect"
|
||||
"strings"
|
||||
@@ -234,7 +235,7 @@ func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt(
|
||||
|
||||
func TestTheSeatsSourceAnswerIsReadAndItsErrorIsAnError(t *testing.T) { // design 47 G §115
|
||||
var asked any
|
||||
read := sourceVia(func(key string, body any) (json.RawMessage, error) {
|
||||
read := sourceVia(context.Background(), func(_ busretry.Open, key string, body any) (json.RawMessage, error) {
|
||||
asked = body
|
||||
if key != "seat:mesh-delivery.source" {
|
||||
t.Errorf("asked %q", key)
|
||||
@@ -243,7 +244,7 @@ func TestTheSeatsSourceAnswerIsReadAndItsErrorIsAnError(t *testing.T) { // desig
|
||||
return json.RawMessage(`{"error":"source.bad is not readable"}`), nil
|
||||
}
|
||||
return json.RawMessage(`{"branch":"main","revision":7,"repositories":[{"repository":"novox/mesh-delivery","head":"` + head + `","trunk":"main"}]}`), nil
|
||||
})
|
||||
}, func(context.Context, string) (bool, error) { return true, nil }, t.Logf)
|
||||
a, err := read("main")
|
||||
if err != nil || a.Revision != 7 || len(a.Repositories) != 1 || asked.(meshdelivery.SourceInput).Branch != "main" {
|
||||
t.Errorf("source = %+v, %v", a, err)
|
||||
|
||||
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
@@ -12,17 +13,19 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// sourceInvoke is the seat mesh-delivery's verb source as the manifest's invokes names it: a watched branch's source
|
||||
// revision, asked of the seat and never read from the gate's state (design 47 G §15, §115; ADR 0596 §2).
|
||||
const sourceInvoke = "seat:" + meshdelivery.Seat + "." + meshdelivery.VerbSource
|
||||
|
||||
// sourceVia asks the seat for one branch's source revision through call (process.Conn.CallKept, a read) and reads its
|
||||
// sourceVia asks the seat for one branch's source revision through call (process.Conn.CallKeptWhileOpen, a read asked again while the branch is watched, A-190) and reads its
|
||||
// declared answer; an answer that carries an error is that error.
|
||||
func sourceVia(call toolCaller) func(branch string) (meshdelivery.SourceAnswer, error) {
|
||||
func sourceVia(ctx context.Context, call openCaller, watched func(ctx context.Context, branch string) (bool, error), logf func(string, ...any)) func(branch string) (meshdelivery.SourceAnswer, error) {
|
||||
return func(branch string) (meshdelivery.SourceAnswer, error) {
|
||||
raw, err := call(sourceInvoke, meshdelivery.SourceInput{Branch: branch})
|
||||
raw, err := keptWhileOpen(ctx, call, logf, "source."+branch, sourceInvoke, meshdelivery.SourceInput{Branch: branch},
|
||||
func(ctx context.Context) (bool, error) { return watched(ctx, branch) })
|
||||
if err != nil {
|
||||
return meshdelivery.SourceAnswer{}, err
|
||||
}
|
||||
@@ -97,6 +100,9 @@ func (s *sourceFollow) moved(ctx context.Context, branch string) {
|
||||
// through the take-in, another watched branch's into its own revision (A-93).
|
||||
func (s *sourceFollow) take(ctx context.Context, branch string) {
|
||||
a, err := s.read(branch)
|
||||
if errors.Is(err, index.ErrAskClosed) { // the branch is no longer watched: not said unanswered (A-190)
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
s.logf("source.%s is not read, and is read again at its next source-moved: %v", branch, err)
|
||||
return
|
||||
|
||||
@@ -10,12 +10,13 @@ import (
|
||||
"mesh-delivery-indexer/internal/index"
|
||||
)
|
||||
|
||||
// stepsVia asks a build seat's verb steps through call (process.Conn.CallKept, ADR 0668 §1) and reads its answer (design
|
||||
// stepsVia asks a build seat's verb steps through call (process.Conn.CallKeptWhileOpen, ADR 0668 §1, asked again while the candidate is held, A-190) and reads its answer (design
|
||||
// 47 G §704); the seats' verbs have one form, so nodebuildgo's contract reads each seat's answer.
|
||||
func stepsVia(call toolCaller) index.StepsCaller {
|
||||
return func(_ context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) {
|
||||
func stepsVia(call openCaller, held func(ctx context.Context, module, fingerprint string) (bool, error), logf func(string, ...any)) index.StepsCaller {
|
||||
return func(ctx context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) {
|
||||
verb := "seat:" + seat + "." + index.StepsVerb
|
||||
raw, err := call(verb, in)
|
||||
raw, err := keptWhileOpen(ctx, call, logf, "steps."+seat+"."+in.Module+"."+in.Fingerprint, verb, in,
|
||||
func(ctx context.Context) (bool, error) { return held(ctx, in.Module, in.Fingerprint) })
|
||||
if err != nil {
|
||||
return nodebuildgo.StepsAnswer{}, err
|
||||
}
|
||||
|
||||
@@ -136,11 +136,12 @@ func sayVersionsFilled(conn *nats.Conn, cred process.Account) func(context.Conte
|
||||
}
|
||||
}
|
||||
|
||||
// versionsVia asks the builder's tool builder_versions for one name as a kept call (ADR 0668 §1, §2) and reads its answer.
|
||||
func versionsVia(call toolCaller) index.VersionsAsker {
|
||||
return func(_ context.Context, name string) (index.BuilderVersionsAnswer, error) {
|
||||
// versionsVia asks the builder's tool builder_versions for one name as a kept call (ADR 0668 §1, §2) asked again while the module is in the newest view (A-190) and reads its answer.
|
||||
func versionsVia(call openCaller, inView func(ctx context.Context, module string) (bool, error), logf func(string, ...any)) index.VersionsAsker {
|
||||
return func(ctx context.Context, module, name string) (index.BuilderVersionsAnswer, error) {
|
||||
var a index.BuilderVersionsAnswer
|
||||
raw, err := call(index.BuilderVersionsCall, map[string]any{"name": name})
|
||||
raw, err := keptWhileOpen(ctx, call, logf, "versions."+name, index.BuilderVersionsCall, map[string]any{"name": name},
|
||||
func(ctx context.Context) (bool, error) { return inView(ctx, module) })
|
||||
if err != nil {
|
||||
return a, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
// ErrAskClosed is what a kept call answers when its ask was closed while it was made again (design 47 G §351, §460, A-190):
|
||||
// what the ask served no longer stands, so the call stopped and is not said unanswered.
|
||||
var ErrAskClosed = errors.New("the ask is closed: what it served no longer stands")
|
||||
|
||||
// takingIn marks a context as the take-in of one commit's merged message, which is pending while that context lives.
|
||||
type takingIn struct{ repository, merge string }
|
||||
|
||||
type takingInKey struct{}
|
||||
|
||||
// forTakeIn is ctx marked as the take-in of the merged message of repository's commit merge.
|
||||
func forTakeIn(ctx context.Context, repository, merge string) context.Context {
|
||||
return context.WithValue(ctx, takingInKey{}, takingIn{repository, merge})
|
||||
}
|
||||
|
||||
// TakeInPending says the take-in of the commit merge of repository is still pending (A-190, for the forge bridge's
|
||||
// merged_message): its take-in is the one ctx is made for, or a pending entry names it.
|
||||
func (ix *Indexer) TakeInPending(ctx context.Context, repository, merge string) (bool, error) {
|
||||
if ctx.Err() != nil {
|
||||
return false, nil
|
||||
}
|
||||
if t, ok := ctx.Value(takingInKey{}).(takingIn); ok && t.repository == repository && t.merge == merge {
|
||||
return true, nil
|
||||
}
|
||||
if ix.Pending == nil {
|
||||
return false, nil
|
||||
}
|
||||
entries, err := ReadPending(ctx, ix.Pending, ix.logf)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return slices.ContainsFunc(entries, func(e PendingEntry) bool {
|
||||
return e.Forge.Merge == merge && e.Forge.Repository == repository
|
||||
}), nil
|
||||
}
|
||||
|
||||
// SeedEntryHeld says the newest trunk revision still holds the entry of module at the fingerprint fp, the candidate the
|
||||
// seed asks steps for (A-190).
|
||||
func (ix *Indexer) SeedEntryHeld(ctx context.Context, module, fp string) (bool, error) {
|
||||
if ctx.Err() != nil || ix.Revisions == nil {
|
||||
return false, ctx.Err()
|
||||
}
|
||||
rev, found, err := newestTrunk(ctx, ix.Revisions)
|
||||
if err != nil || !found {
|
||||
return false, err
|
||||
}
|
||||
_, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return slices.ContainsFunc(entries, func(e meshcatalog.CatalogEntry) bool { return e.Module == module && e.Fp == fp }), nil
|
||||
}
|
||||
|
||||
// ModuleInNewestView says module is among the entries of the newest trunk revision, the view the versions are filled for (A-190).
|
||||
func (ix *Indexer) ModuleInNewestView(ctx context.Context, module string) (bool, error) {
|
||||
if ctx.Err() != nil || ix.Revisions == nil {
|
||||
return false, ctx.Err()
|
||||
}
|
||||
rev, found, err := newestTrunk(ctx, ix.Revisions)
|
||||
if err != nil || !found {
|
||||
return false, err
|
||||
}
|
||||
_, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return slices.ContainsFunc(entries, func(e meshcatalog.CatalogEntry) bool { return e.Module == module }), nil
|
||||
}
|
||||
|
||||
// BranchWatched says branch (as the source's key names it, escaped; empty asks the list of every branch) is watched: the
|
||||
// trunk always, another branch while a registration names it (A-190). With no registry, every branch is.
|
||||
func (ix *Indexer) BranchWatched(ctx context.Context, branch string) (bool, error) {
|
||||
if ctx.Err() != nil {
|
||||
return false, nil
|
||||
}
|
||||
if isTrunkBranch(branch) || ix.Registry == nil || ix.Registry.Registrations == nil {
|
||||
return true, nil
|
||||
}
|
||||
name, err := keys.UnescapeName(branch)
|
||||
if err != nil {
|
||||
return false, nil
|
||||
}
|
||||
regs, err := ix.Registry.RegistrationsAnswer(ctx, meshcatalog.RegistrationsInput{})
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
for _, r := range regs.Registrations {
|
||||
if slices.Contains(r.Branches, name) {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// A-190: a kept call is asked again only while what it serves still stands; each of the indexer's four has its predicate.
|
||||
func TestKeptCallsOpenWhileTheirAskServesSomething(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
t.Run("the take-in of a commit is pending while an entry names it or the take-in is made", func(t *testing.T) {
|
||||
pending := newMem()
|
||||
val, _ := encodePending(Pending{Seq: 1, Ref: &PendingRef{Repository: "novox/r", Merge: "c1"}})
|
||||
if err := pending.Create(ctx, "m.1", val); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ix := &Indexer{Pending: pending}
|
||||
cancelled, cancel := context.WithCancel(ctx)
|
||||
cancel()
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
ctx context.Context
|
||||
repo, commit string
|
||||
want bool
|
||||
}{
|
||||
{"held", ctx, "novox/r", "c1", true},
|
||||
{"taken in or dropped", ctx, "novox/r", "c2", false},
|
||||
{"in hand", forTakeIn(ctx, "novox/r", "c2"), "novox/r", "c2", true},
|
||||
{"in hand for another commit", forTakeIn(ctx, "novox/r", "c3"), "novox/r", "c2", false},
|
||||
{"the take-in's context ended", cancelled, "novox/r", "c1", false},
|
||||
} {
|
||||
if got, err := ix.TakeInPending(c.ctx, c.repo, c.commit); err != nil || got != c.want {
|
||||
t.Errorf("%s: pending = %v, %v, want %v", c.name, got, err, c.want)
|
||||
}
|
||||
}
|
||||
})
|
||||
t.Run("the seed's candidate and the versions' module are in the newest trunk revision", func(t *testing.T) {
|
||||
ix := &Indexer{Revisions: newMemObjects(1 << 20)}
|
||||
keep := func(n int64, fp string) {
|
||||
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: n, Kind: "trunk"}, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Fp: fp, Manifest: map[string]any{}}})
|
||||
if err == nil {
|
||||
err = KeepRevision(ctx, ix.Revisions, rev, objects)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
keep(1, "fp1")
|
||||
if held, err := ix.SeedEntryHeld(ctx, "avahi", "fp1"); err != nil || !held {
|
||||
t.Errorf("the entry at its fingerprint: %v, %v", held, err)
|
||||
}
|
||||
if in, err := ix.ModuleInNewestView(ctx, "avahi"); err != nil || !in {
|
||||
t.Errorf("the module in the view: %v, %v", in, err)
|
||||
}
|
||||
if in, _ := ix.ModuleInNewestView(ctx, "gone"); in {
|
||||
t.Error("a module the view does not hold is in it")
|
||||
}
|
||||
keep(2, "fp2")
|
||||
if held, _ := ix.SeedEntryHeld(ctx, "avahi", "fp1"); held {
|
||||
t.Error("a candidate the newest revision replaced is still held")
|
||||
}
|
||||
})
|
||||
t.Run("a branch is watched while a registration names it", func(t *testing.T) {
|
||||
g := newRegRig()
|
||||
ix := &Indexer{Registry: g.r}
|
||||
raw, _ := json.Marshal(meshcatalog.Registration{Module: "avahi", Branches: []string{"main", "feat"}})
|
||||
if err := g.r.Registrations.Create(ctx, "avahi", raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if w, err := ix.BranchWatched(ctx, TrunkBranch); err != nil || !w {
|
||||
t.Errorf("the trunk: %v, %v", w, err)
|
||||
}
|
||||
if w, _ := ix.BranchWatched(ctx, "feat"); !w {
|
||||
t.Error("a branch a registration names is not watched")
|
||||
}
|
||||
if w, _ := ix.BranchWatched(ctx, "other"); w {
|
||||
t.Error("a branch no registration names is watched")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A-190: a call that ends with its ask closed is not said unanswered: no finding, and the fill and the seed go on.
|
||||
func TestAClosedAskIsNotUnanswered(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
closed := fmt.Errorf("steps: %w", ErrAskClosed)
|
||||
ix := &Indexer{Findings: newFindings(), Heard: newMem(), Revisions: newMemObjects(1 << 20), Versions: newMem()}
|
||||
rev, objects, _ := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 1, Kind: "trunk"}, map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Fp: "fp", Manifest: map[string]any{},
|
||||
Artifacts: []meshcatalog.Artifact{{Ref: "avahi:program", Language: "go"}}}})
|
||||
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defs := map[string]seatdef.Definition{"node-build-go": {Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: "go"}}}}
|
||||
if _, err := ix.SeedModulesFrom(ctx, defs, func(context.Context, string, nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) {
|
||||
return nodebuildgo.StepsAnswer{}, closed
|
||||
}); err != nil || len(ix.stepsSince) != 0 {
|
||||
t.Errorf("a closed steps ask counted unanswered: %v %v", err, ix.stepsSince)
|
||||
}
|
||||
if err := ix.FillVersions(ctx, func(context.Context, string, string) (BuilderVersionsAnswer, error) {
|
||||
return BuilderVersionsAnswer{}, fmt.Errorf("builder_versions: %w", ErrAskClosed)
|
||||
}); errors.Is(err, ErrAskClosed) || (err != nil && strings.Contains(err.Error(), "is not answered")) {
|
||||
t.Errorf("a closed versions ask is unanswered: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -1490,7 +1490,7 @@ func (ix *Indexer) TakeInMerge(ctx context.Context, repository, merge string, se
|
||||
r.Answer = "this indexer does not ask the forge bridge"
|
||||
return r
|
||||
}
|
||||
data, err := ix.Message(ctx, repository, merge, 0)
|
||||
data, err := ix.Message(forTakeIn(ctx, repository, merge), repository, merge, 0)
|
||||
if err != nil {
|
||||
r.Answer = fmt.Sprintf("the forge bridge could not answer for the merge of %s at %s: %v", repository, merge, err)
|
||||
return r
|
||||
|
||||
@@ -2,6 +2,7 @@ package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"path"
|
||||
@@ -91,6 +92,7 @@ func (ix *Indexer) SeedModulesFrom(ctx context.Context, defs map[string]seatdef.
|
||||
Fetch: &nodebuildgo.Fetch{Repository: e.Repository, Commit: e.Commit, Dir: e.Path, Paths: namedPaths(e, name), Packages: maps.Clone(e.Packages)}}
|
||||
ans, err := steps(ctx, seat, in)
|
||||
switch {
|
||||
case errors.Is(err, ErrAskClosed): // the entry is no longer held: not said unanswered (A-190)
|
||||
case err != nil:
|
||||
unanswered[seat] = err
|
||||
case ans.Error != "":
|
||||
|
||||
@@ -216,7 +216,7 @@ func (ix *Indexer) CatchUp(ctx context.Context) TakenIn {
|
||||
|
||||
// mergedAt is the forge bridge's merged message of a repository's head.
|
||||
func (ix *Indexer) mergedAt(ctx context.Context, repository, head string) (forgemsg.Forge, error) {
|
||||
data, err := ix.Message(ctx, repository, head, 0)
|
||||
data, err := ix.Message(forTakeIn(ctx, repository, head), repository, head, 0)
|
||||
if err != nil {
|
||||
return forgemsg.Forge{}, fmt.Errorf("the forge bridge could not answer: %w", err)
|
||||
}
|
||||
|
||||
@@ -46,7 +46,7 @@ type BuilderVersionsAnswer struct {
|
||||
}
|
||||
|
||||
// VersionsAsker asks builder_versions for a name's versions (a kept call, ADR 0668 §1, §2).
|
||||
type VersionsAsker func(ctx context.Context, name string) (BuilderVersionsAnswer, error)
|
||||
type VersionsAsker func(ctx context.Context, module, name string) (BuilderVersionsAnswer, error)
|
||||
|
||||
// versionNameOf is the name the builder numbers an artifact under: its package's name in its ecosystem, else
|
||||
// <module>/<artifact> (images, bundles and archives).
|
||||
@@ -97,7 +97,10 @@ func (ix *Indexer) FillVersions(ctx context.Context, ask VersionsAsker) error {
|
||||
per := []found1{}
|
||||
for _, a := range e.Artifacts {
|
||||
name := versionNameOf(e.Module, a)
|
||||
ans, err := ask(ctx, name)
|
||||
ans, err := ask(ctx, e.Module, name)
|
||||
if errors.Is(err, ErrAskClosed) {
|
||||
continue // the module left the view: nothing waits on its versions (A-190)
|
||||
}
|
||||
if err != nil {
|
||||
unanswered = err
|
||||
continue
|
||||
|
||||
@@ -58,7 +58,7 @@ func (ix *Indexer) filledEdges(ctx context.Context, entry meshcatalog.CatalogEnt
|
||||
}
|
||||
ans, ok := answers[name]
|
||||
if !ok {
|
||||
a, err := ask(ctx, name)
|
||||
a, err := ask(ctx, module, name)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -115,7 +115,7 @@ func TestTheFirstFillOfTheVersionsComesFromTheBuildersPublished(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
up := false
|
||||
ask := func(_ context.Context, name string) (BuilderVersionsAnswer, error) {
|
||||
ask := func(_ context.Context, _, name string) (BuilderVersionsAnswer, error) {
|
||||
if !up {
|
||||
return BuilderVersionsAnswer{}, errors.New("no responders")
|
||||
}
|
||||
@@ -457,7 +457,7 @@ func TestTheFirstFillGivesTheNewestVersionsTheirFilledEdges(t *testing.T) {
|
||||
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ask := func(_ context.Context, name string) (BuilderVersionsAnswer, error) {
|
||||
ask := func(_ context.Context, _, name string) (BuilderVersionsAnswer, error) {
|
||||
pub := func(v, d string, at time.Time) BuilderVersion {
|
||||
return BuilderVersion{Version: v, State: "published", Digest: d, Revision: "9", At: at}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user