Merge branch 'feat/47-G-build' into feat/47-G-build-teller
This commit is contained in:
@@ -493,9 +493,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
if kv := kvs[writtenBucket]; kv != nil {
|
||||
// The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow
|
||||
// once the first registrations are written (§95, §353).
|
||||
p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf)
|
||||
if err := hearEnds(hub, p.heard); err != nil {
|
||||
logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err)
|
||||
p.ends = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf)
|
||||
err := hearEnds(hub, p.ends)
|
||||
if err == nil && bus != nil {
|
||||
err = followRegistrations(ctx, hub, bus.Call, p.ends)
|
||||
}
|
||||
if err != nil {
|
||||
logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err)
|
||||
}
|
||||
}
|
||||
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
|
||||
|
||||
@@ -58,10 +58,8 @@ type passer struct {
|
||||
// neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the
|
||||
// first pass after a start, which clears what a previous process may have raised (startClears).
|
||||
neither map[string]neitherOne
|
||||
// ends are the pipeline ends the wants follow (pipelines.go); nil is the trunk-only want, kept until the first registrations.
|
||||
// ends are the pipeline ends the wants follow (pipelines.go), once the first registrations are heard; until then, and nil, the trunk-only want.
|
||||
ends *pipelineEnds
|
||||
// heard is the same follower while the wants do not follow it yet (ends is nil).
|
||||
heard *pipelineEnds
|
||||
// acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket).
|
||||
acts *putActs
|
||||
// noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is
|
||||
|
||||
@@ -5,13 +5,19 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient"
|
||||
mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
@@ -81,6 +87,9 @@ type endStore interface {
|
||||
get(key string) (value []byte, revision uint64, err error)
|
||||
// swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not.
|
||||
swap(key string, value []byte, revision uint64) error
|
||||
// keys are the keys that stand, and drop deletes one.
|
||||
keys() ([]string, error)
|
||||
drop(key string) error
|
||||
}
|
||||
|
||||
var errSwapped = errors.New("the key changed")
|
||||
@@ -115,6 +124,16 @@ func (k kvEnds) swap(key string, value []byte, revision uint64) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (k kvEnds) keys() ([]string, error) {
|
||||
ks, err := k.kv.Keys(k.ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil, nil
|
||||
}
|
||||
return ks, err
|
||||
}
|
||||
|
||||
func (k kvEnds) drop(key string) error { return k.kv.Delete(k.ctx, key) }
|
||||
|
||||
func isWrongSequence(err error) bool {
|
||||
var ae *jetstream.APIError
|
||||
return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
|
||||
@@ -126,7 +145,9 @@ type pipelineEnds struct {
|
||||
st endStore
|
||||
wake func()
|
||||
logf func(string, ...any)
|
||||
mu sync.Mutex
|
||||
// on is whether the wants follow the ends: set by the first registration heard, never before (design 47 G §95, §353).
|
||||
on atomic.Bool
|
||||
mu sync.Mutex
|
||||
// firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design
|
||||
// 47 G §14).
|
||||
firsts map[string]string
|
||||
@@ -278,7 +299,7 @@ func writtenTools(st store) []stdio.Tool {
|
||||
|
||||
// branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one.
|
||||
func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string {
|
||||
if e == nil {
|
||||
if !e.following() {
|
||||
return nil
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
@@ -300,7 +321,7 @@ func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]st
|
||||
// mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already
|
||||
// (a node placed later, a module taken away).
|
||||
func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool {
|
||||
if e == nil {
|
||||
if !e.following() {
|
||||
return true
|
||||
}
|
||||
for _, w := range wanted {
|
||||
@@ -318,10 +339,77 @@ func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[
|
||||
|
||||
// wrote marks the end the wants of module were written from.
|
||||
func (e *pipelineEnds) wrote(module, fp string, branches []string) {
|
||||
if e == nil {
|
||||
if !e.following() {
|
||||
return
|
||||
}
|
||||
for _, b := range branches {
|
||||
e.markWritten(module, b, fp)
|
||||
}
|
||||
}
|
||||
|
||||
// following says whether the wants follow the pipeline ends yet; a nil follower never does.
|
||||
func (e *pipelineEnds) following() bool { return e != nil && e.on.Load() }
|
||||
|
||||
// registered takes a module's registration (design 47 G §353): the first one heard switches the wants to the ends, its first
|
||||
// watched branch is the default a placement runs, and the ends kept for a branch it no longer watches are deleted (§352). gone
|
||||
// is a registration removed: all its ends go.
|
||||
func (e *pipelineEnds) registered(reg mpcat.Registration, gone bool) {
|
||||
e.mu.Lock()
|
||||
if gone || len(reg.Branches) == 0 {
|
||||
delete(e.firsts, reg.Module)
|
||||
} else {
|
||||
e.firsts[reg.Module] = reg.Branches[0]
|
||||
}
|
||||
e.mu.Unlock()
|
||||
kept, err := e.st.keys()
|
||||
if err != nil {
|
||||
e.logf("the ends kept cannot be listed: %v", err)
|
||||
}
|
||||
prefix := keys.EscapeName(reg.Module) + "."
|
||||
for _, k := range kept {
|
||||
rest, ok := strings.CutPrefix(k, prefix)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
branch, err := keys.UnescapeName(rest)
|
||||
if err == nil && !gone && slices.Contains(reg.Branches, branch) {
|
||||
continue
|
||||
}
|
||||
if err := e.st.drop(k); err != nil {
|
||||
e.logf("the end %s is not deleted: %v", k, err)
|
||||
}
|
||||
}
|
||||
if !e.on.Swap(true) {
|
||||
e.logf("the first registration is heard (%s): the wants follow the pipelines' ends from now", reg.Module)
|
||||
}
|
||||
e.wake()
|
||||
}
|
||||
|
||||
// followRegistrations follows the registrations through the seat mesh-catalog (design 47 G §353): a first view of its verb
|
||||
// and each registration-changed on the hub. It returns when the hub's handler is registered; the first view is asked on its own
|
||||
// goroutine, the wants keep the trunk-only form until a registration is heard.
|
||||
func followRegistrations(ctx context.Context, hub *process.EventHub, call mcc.Call, e *pipelineEnds) error {
|
||||
regs := mcc.NewRegistrations(call)
|
||||
regs.Follower.Logf = e.logf
|
||||
regs.Follower.Changed = func(module string, value json.RawMessage, gone bool, _ time.Time) {
|
||||
reg := mpcat.Registration{Module: module}
|
||||
if !gone {
|
||||
_ = json.Unmarshal(value, ®)
|
||||
}
|
||||
e.registered(reg, gone)
|
||||
}
|
||||
if err := hub.Handle(mpcat.SubjectRegistrationChanged, func(m jetstream.Msg) {
|
||||
if err := regs.Hear(ctx, m.Data()); err != nil {
|
||||
e.logf("%s: %v", m.Subject(), err)
|
||||
}
|
||||
_ = m.Ack()
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
if err := regs.Start(ctx); err != nil && ctx.Err() == nil {
|
||||
e.logf("the first view of the registrations is not answered: %v", err)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
@@ -34,6 +35,13 @@ type memEnds struct {
|
||||
func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} }
|
||||
|
||||
func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil }
|
||||
func (m *memEnds) keys() (out []string, _ error) {
|
||||
for k := range m.vals {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (m *memEnds) drop(key string) error { delete(m.vals, key); delete(m.revs, key); return nil }
|
||||
func (m *memEnds) swap(key string, v []byte, rev uint64) error {
|
||||
if m.revs[key] != rev {
|
||||
return errSwapped
|
||||
@@ -79,6 +87,7 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) {
|
||||
} {
|
||||
woken := 0
|
||||
e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf)
|
||||
e.on.Store(true)
|
||||
for _, p := range c.ends {
|
||||
if p.Group != "" {
|
||||
continue // heard and ignored, as hearPipeline does
|
||||
@@ -100,12 +109,14 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) {
|
||||
}
|
||||
// A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it).
|
||||
e := newPipelineEnds(newMemEnds(), func() {}, t.Logf)
|
||||
e.on.Store(true)
|
||||
if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) {
|
||||
t.Error("a want naming the fingerprint was held")
|
||||
}
|
||||
// markWritten keeps the end and says it was written from; a newer end replaces it unwritten.
|
||||
st := newMemEnds()
|
||||
e = newPipelineEnds(st, func() {}, t.Logf)
|
||||
e.on.Store(true)
|
||||
_ = e.take(pipe("web", "main", 5, 1, fp1, pass), true)
|
||||
e.wrote("web", fp1, []string{"main"})
|
||||
var end End
|
||||
@@ -153,6 +164,7 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
|
||||
p.w.put(buckets.Presence, "g14", on)
|
||||
if c.follow {
|
||||
p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf)
|
||||
p.ends.on.Store(true)
|
||||
if c.passed {
|
||||
_ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true)
|
||||
}
|
||||
@@ -172,3 +184,30 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §95, §130, §353, §352 (test 100): the wants keep the trunk-only form until the first registration is heard and
|
||||
// switch on it; a branch a registration no longer watches loses its end, and its first branch is a placement's default.
|
||||
func TestTheWantsSwitchOnTheFirstRegistration(t *testing.T) {
|
||||
st := newMemEnds()
|
||||
woken := 0
|
||||
e := newPipelineEnds(st, func() { woken++ }, t.Logf)
|
||||
_ = e.take(pipe("web", "main", 5, 1, "fp1", mpr.PipelineEndedOutcomeFailed), false)
|
||||
_ = e.take(pipe("web", "beta", 5, 1, "fp1", mpr.PipelineEndedOutcomePassed), true)
|
||||
if e.following() || !e.mayWant("web", "fp1", []string{"main"}, nil) {
|
||||
t.Fatal("the follower held a want before any registration")
|
||||
}
|
||||
e.registered(mpcat.Registration{Module: "web", Branches: []string{"beta"}}, false)
|
||||
if !e.following() || woken == 0 {
|
||||
t.Error("the first registration did not switch the wants on")
|
||||
}
|
||||
if e.mayWant("web", "fp1", []string{"main"}, nil) || !e.mayWant("web", "fp1", []string{"beta"}, nil) {
|
||||
t.Error("after the switch the wants did not follow the ends")
|
||||
}
|
||||
if len(st.vals) != 1 || e.branchOf("web", "") != "beta" || e.branchOf("other", "") != trunkBranch {
|
||||
t.Errorf("ends kept %d, default branch %q: an unwatched branch's end stayed, or the default is wrong", len(st.vals), e.branchOf("web", ""))
|
||||
}
|
||||
e.registered(mpcat.Registration{Module: "web"}, true)
|
||||
if len(st.vals) != 0 {
|
||||
t.Errorf("a registration gone left %d ends", len(st.vals))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -123,11 +123,11 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
|
||||
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) {
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) {
|
||||
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes)
|
||||
}
|
||||
if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
"seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||
"seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||
"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
|
||||
"seat:node-backup.backed-up", "seat:node-backup.now",
|
||||
"seat:node-engine.manifest-reads", // ADR 0618 §3
|
||||
|
||||
@@ -39,6 +39,7 @@
|
||||
"operator-channel"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-catalog.registration-changed",
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
"mesh-catalog.seat-definition-removed",
|
||||
"mesh-delivery-indexer.views.graph",
|
||||
@@ -53,6 +54,7 @@
|
||||
"mesh-delivery-indexer.graph",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"seat:mesh-catalog.registrations",
|
||||
"seat:mesh-catalog.seat-definitions",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-hand-acts.hand-acts",
|
||||
|
||||
@@ -591,6 +591,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := ix.Reconcile(ctx); err != nil {
|
||||
logger.Printf("dropping the modules the trunk does not hold at start: %v", err)
|
||||
}
|
||||
// Every contracts module's trunk definitions read before any manifest is judged and before the forge bridge's stream is
|
||||
// followed (design 47, the indexer reads every definition before it judges): a definition merged while it was down, or
|
||||
// one no event announced, is never missing from the judging below. Once the state is bound, so on its first binding too.
|
||||
// Through the seat git (milestone 47/G W4.15, design 47 *A part that cannot read the forge keeps what it has*): a seat that
|
||||
// does not answer is waited for at most index.GitWaitBound before the judging below runs on the kept definitions, said in
|
||||
// indexer_first_views and as view-waits past five minutes, and read again on the next forge message heard.
|
||||
defs := gitWaits(ctx, ix, findings, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views does not say the wait for the seat git", logger.Printf), logger.Printf)
|
||||
seatDefs.Ready(func(jetstream.KeyValue) { defs.Start(ctx) })
|
||||
// Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578
|
||||
// §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage).
|
||||
verdicts.Ready(func(jetstream.KeyValue) {
|
||||
@@ -603,13 +611,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}()
|
||||
// A pass that fails (a read the bus did not answer) is judged again at the next message or source revision the
|
||||
// indexer hears (index.kickJudge): no timer.
|
||||
j, err := ix.JudgeAtStart(ctx)
|
||||
if err != nil {
|
||||
logger.Printf("judging the manifests at start: %v; judged again at the next message heard", err)
|
||||
judge := func(fromKept bool) {
|
||||
j, err := ix.JudgeAtStart(ctx)
|
||||
if err != nil {
|
||||
logger.Printf("judging the manifests at start (from kept definitions: %v): %v; judged again at the next message heard", fromKept, err)
|
||||
return
|
||||
}
|
||||
logger.Printf("judged %d manifests at start (from kept definitions: %v): %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v",
|
||||
j.Judged, fromKept, j.Written, len(j.Refused), j.Outgrown, j.Waiting)
|
||||
}
|
||||
if !defs.Waiting() {
|
||||
judge(false)
|
||||
return
|
||||
}
|
||||
logger.Printf("judged %d manifests at start: %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v",
|
||||
j.Judged, j.Written, len(j.Refused), j.Outgrown, j.Waiting)
|
||||
// The seat git did not answer: the judging waits for it at most index.GitWaitBound, off the start, so nothing else of
|
||||
// the start is held behind it.
|
||||
go func() { judge(defs.Ready(ctx)) }()
|
||||
})
|
||||
// A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are
|
||||
// made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again
|
||||
@@ -736,6 +753,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
handle(ctx, m, ix, heard, logger)
|
||||
ix.KickFill(ctx)
|
||||
unreadAgain(ctx, ix, logger)
|
||||
defs.Heard(ctx)
|
||||
}, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@@ -180,3 +181,27 @@ func followAssignments(ctx context.Context, hub *process.EventHub, cred process.
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// gitWaits is the indexer's read of the trunk definitions at its start through the seat git (index.DefinitionsAtStart): its
|
||||
// wait said as a first view of the seat's (Owner git, State seat-definitions) in the indexer's first views, and its finding
|
||||
// view-waits written past views.WaitsBound and cleared once the seat answers.
|
||||
func gitWaits(ctx context.Context, ix *index.Indexer, findings, status *process.AwaitedBucket, logf func(string, ...any)) *index.DefinitionsAtStart {
|
||||
finding := func(s views.Status, raised bool) {
|
||||
w := sdkfindings.Writer(ctx, findings.KV)
|
||||
f := views.ViewWaitsFinding(index.Module, s, time.Now())
|
||||
var err error
|
||||
if raised {
|
||||
err = w.Put(f)
|
||||
} else {
|
||||
err = w.Delete(f)
|
||||
}
|
||||
if err != nil {
|
||||
logf("views: the finding view-waits of the seat git is not written: %v", err)
|
||||
}
|
||||
}
|
||||
f := &views.Follower{Owner: git.Seat, Tool: "seat:" + git.Seat + "." + git.VerbTrees, State: index.SeatDefinitionsState, Logf: logf,
|
||||
RaiseWaits: func(s views.Status) { finding(s, true) }, ClearWaits: func(s views.Status) { finding(s, false) }}
|
||||
status.Ready(func(kv jetstream.KeyValue) { f.Said = views.KeepStatus(ctx, kv, logf) })
|
||||
return &index.DefinitionsAtStart{Read: ix.ReadTrunkDefinitions, Waits: f, Logf: logf,
|
||||
Judge: func(ctx context.Context) error { _, err := ix.JudgeAtStart(ctx); return err }}
|
||||
}
|
||||
|
||||
@@ -125,6 +125,36 @@ func TestTheControlRecords(t *testing.T) {
|
||||
// and source, says seat-definition-changed only when a digest moved, keeps the trunk's entry over one that does not
|
||||
// parse (its finding), removes a pull request's entries when it closes, keeps a verdict's out-of-step files, and
|
||||
// raises generated-types-out-of-step from a trunk build end, cleared only by a clean one of the same module.
|
||||
// Milestone 47/G W4.15, design 47 test 54: the seat git not answering at start holds the judging at most its bound, which
|
||||
// then runs on the kept definitions with the wait said; the read once the forge is heard holds the view and judges again.
|
||||
t.Run("W4.15 the start waits for the seat git, judges from kept definitions, and judges again once it answers", func(t *testing.T) {
|
||||
answers := []error{fmt.Errorf("mesh-contracts: %w", errForgeRead), fmt.Errorf("again: %w", errForgeRead), nil}
|
||||
var said []string
|
||||
judged := 0
|
||||
d := &DefinitionsAtStart{Bound: time.Millisecond, Waits: waitsFake{&said},
|
||||
Read: func(context.Context) (int, error) { err := answers[0]; answers = answers[1:]; return 1, err },
|
||||
Judge: func(context.Context) error { judged++; return nil }}
|
||||
ctx := context.Background()
|
||||
d.Start(ctx)
|
||||
if !d.Waiting() || len(said) != 1 || said[0] != "waits "+GitDefinitionsView {
|
||||
t.Fatalf("waiting %v, said %q: want the wait for the seat git said once", d.Waiting(), said)
|
||||
}
|
||||
if !d.Ready(ctx) || judged != 0 {
|
||||
t.Fatalf("the start past its bound: want it judged from the kept definitions, judged again %d", judged)
|
||||
}
|
||||
d.Heard(ctx) // the forge heard, the seat still not answering: the wait stands, said once
|
||||
if !d.Waiting() || len(said) != 1 || judged != 0 {
|
||||
t.Fatalf("a read still not answered: waiting %v, said %q, judged %d", d.Waiting(), said, judged)
|
||||
}
|
||||
d.Heard(ctx)
|
||||
if d.Waiting() || len(said) != 2 || said[1] != "holds "+GitDefinitionsView || judged != 1 || d.Ready(ctx) {
|
||||
t.Errorf("the seat answering: waiting %v, said %q, judged again %d: want the view held and every manifest judged again once", d.Waiting(), said, judged)
|
||||
}
|
||||
d.Heard(ctx) // nothing waits: no read, no judging
|
||||
if judged != 1 || len(answers) != 0 {
|
||||
t.Errorf("a forge message with nothing waiting read or judged again")
|
||||
}
|
||||
})
|
||||
t.Run("seat definitions", func(t *testing.T) {
|
||||
const repo, m1, m2, m3, head = "novox/mesh-contracts", "m1", "m2", "m3", "h1"
|
||||
manifest := `{"module":"mesh-contracts","version":"1.0","build":{"artifacts":[{"name":"go","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}`
|
||||
@@ -315,6 +345,29 @@ func TestTheControlRecords(t *testing.T) {
|
||||
t.Errorf("said %+v: only a trunk's definition is said (ADR 0636 §4)", c)
|
||||
}
|
||||
}
|
||||
// Design 47 (the indexer reads every definition before it judges): at its start the indexer reads every contracts
|
||||
// module's trunk at its repository's head in the source revision, so a definition merged with no event heard is kept
|
||||
// before any manifest is judged; a repository the source names no head of is read at the module's own commit.
|
||||
kept, _ = meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "mesh-contracts", Manifest: manifest})
|
||||
mkey, _ = meshcatalogclient.ManifestsKey(meshcatalogclient.ManifestDigest([]byte(manifest)))
|
||||
ix.Manifests.(*memStore).vals[mkey] = kept
|
||||
mod, _ = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: meshcatalogclient.ManifestDigest([]byte(manifest)), Repository: repo, Commit: m3, Path: meshcatalogclient.RootPath})
|
||||
ix.Modules.(*memStore).vals["mesh-contracts"] = mod
|
||||
files["h3"] = map[string]string{"seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0500", 1)}
|
||||
ix.source = sourceNow{rev: 7, heads: map[string]string{repo: "h3"}}
|
||||
if n, err := ix.ReadTrunkDefinitions(ctx); err != nil || n != 1 {
|
||||
t.Fatalf("read at start = %d, %v; want the one contracts module", n, err)
|
||||
}
|
||||
if e, _ := entry("git.trunk"); e.Commit != "h3" {
|
||||
t.Errorf("git.trunk after the read at start = %+v, want the source's head h3", e)
|
||||
}
|
||||
ix.source = sourceNow{}
|
||||
if _, err := ix.ReadTrunkDefinitions(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if e, _ := entry("git.trunk"); e.Commit != m3 {
|
||||
t.Errorf("git.trunk read with no source head = %+v, want the module's own commit", e)
|
||||
}
|
||||
|
||||
// ADR 0636 §5, D-I3, D-I5: a seat of the mesh's own contracts is taken only from its domain's contracts module; a
|
||||
// move there is a handover in either order of the two trunks; a second definer is refused with a finding naming both.
|
||||
@@ -636,3 +689,9 @@ func (m memSource) Keys() ([]string, error) {
|
||||
}
|
||||
|
||||
func (m memSource) Get(key string) (*views.Entry, bool, error) { e, ok := m[key]; return e, ok, nil }
|
||||
|
||||
// waitsFake records what the indexer says it waits for and holds.
|
||||
type waitsFake struct{ said *[]string }
|
||||
|
||||
func (w waitsFake) WaitFor(_ context.Context, view string) { *w.said = append(*w.said, "waits "+view) }
|
||||
func (w waitsFake) Holds(view string) { *w.said = append(*w.said, "holds "+view) }
|
||||
|
||||
@@ -13,12 +13,14 @@ package index
|
||||
// contracts module.
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"path"
|
||||
"slices"
|
||||
"strconv"
|
||||
@@ -30,6 +32,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/seatdefs"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/trunk"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
@@ -278,6 +281,49 @@ func (ix *Indexer) SeatDefinitionsAgain(ctx context.Context, module string, pull
|
||||
return out
|
||||
}
|
||||
|
||||
// ReadTrunkDefinitions reads, at the start and before any manifest is judged, the seat definitions of every contracts
|
||||
// module the trunk holds, at its repository's head in the source revision (at the module's own commit where the source
|
||||
// names none), and keeps them as a merge does (design 47, *The indexer reads every definition before it judges*): a
|
||||
// definition merged while the indexer was down, or one no event announced, is kept before a manifest claiming its seat
|
||||
// is judged. It answers how many contracts modules were read; one the forge did not answer is in the error, and is read
|
||||
// again at its next merged message or by indexer_seat_definitions_again.
|
||||
func (ix *Indexer) ReadTrunkDefinitions(ctx context.Context) (int, error) {
|
||||
if ix.SeatDefinitions == nil || ix.ReadFile == nil || ix.Modules == nil || ix.Manifests == nil {
|
||||
return 0, nil
|
||||
}
|
||||
held, err := ix.heldModules(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ix.mu.Lock()
|
||||
heads := maps.Clone(ix.source.heads)
|
||||
ix.mu.Unlock()
|
||||
read := 0
|
||||
var errs []error
|
||||
for _, name := range slices.Sorted(maps.Keys(held)) {
|
||||
m := held[name]
|
||||
manifest, _, err := ix.manifestRaw(ctx, m.Manifest)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s's manifest: %w", name, err))
|
||||
continue
|
||||
}
|
||||
dir := cmp.Or(m.Path, meshcatalogclient.RootPath)
|
||||
commit := m.Commit
|
||||
for r, head := range heads {
|
||||
if trunk.SameRepository(r, m.Repository) && head != "" {
|
||||
commit = head
|
||||
}
|
||||
}
|
||||
for _, c := range contractsModules(map[string]string{dir: manifest}) {
|
||||
read++
|
||||
if err := ix.keepTrunk(ctx, m.Repository, commit, c); err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s at %.12s: %w", name, commit, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
return read, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// keepTrunk writes a contracts module's trunk definitions at a merge: each that parses where its digest moved (said),
|
||||
// each the module's index no longer names removed (said with no digest), and the module's unparsed finding written or
|
||||
// cleared. A definition that does not parse leaves the trunk's entry as it stands.
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// GitWaitBound is how long the indexer waits, at its start, for the seat git to answer the read of every contracts module's
|
||||
// trunk definitions before it judges from the definitions it kept (design 47, *A part that cannot read the forge keeps
|
||||
// what it has, and says so*): longer than a forge's restart, short enough that a start is not held behind a forge down
|
||||
// for good.
|
||||
const GitWaitBound = 2 * time.Minute
|
||||
|
||||
// GitDefinitionsView is what the indexer says it waits for in its first views while the seat git does not answer.
|
||||
const GitDefinitionsView = "trunk seat definitions"
|
||||
|
||||
// ViewWaiter is where the indexer says it waits for the seat git and since when, and raises view-waits past its bound
|
||||
// (mesh-sdk views.Follower's WaitFor and Holds).
|
||||
type ViewWaiter interface {
|
||||
WaitFor(ctx context.Context, view string)
|
||||
Holds(view string)
|
||||
}
|
||||
|
||||
// DefinitionsAtStart is the indexer's read of the trunk definitions at its start through the seat git: the first read,
|
||||
// a wait of at most Bound for the seat when it does not answer, the judging from the kept definitions after it, and the
|
||||
// read again once the seat answers, which judges again every manifest judged from kept definitions.
|
||||
type DefinitionsAtStart struct {
|
||||
// Read reads every contracts module's trunk definitions (Indexer.ReadTrunkDefinitions).
|
||||
Read func(ctx context.Context) (int, error)
|
||||
// Judge judges every manifest again, as at the start (Indexer.JudgeAtStart).
|
||||
Judge func(ctx context.Context) error
|
||||
Waits ViewWaiter
|
||||
Bound time.Duration
|
||||
Logf func(string, ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
answered chan struct{} // closed once a read is answered
|
||||
waiting bool // a read was not answered and no read since was
|
||||
fromKept bool // the judging at start ran on kept definitions
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) init() {
|
||||
if d.answered == nil {
|
||||
d.answered = make(chan struct{})
|
||||
}
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) logf(format string, args ...any) {
|
||||
if d.Logf != nil {
|
||||
d.Logf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// Start reads the trunk definitions once; a seat that does not answer is said as a wait (Waits.WaitFor) and read again on
|
||||
// the next sign that the forge runs (Heard). A read answered with other errors (a manifest it cannot read) is answered.
|
||||
func (d *DefinitionsAtStart) Start(ctx context.Context) {
|
||||
d.mu.Lock()
|
||||
d.init()
|
||||
d.mu.Unlock()
|
||||
d.read(ctx, "at start")
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) read(ctx context.Context, when string) bool {
|
||||
n, err := d.Read(ctx)
|
||||
if errors.Is(err, errForgeRead) {
|
||||
d.mu.Lock()
|
||||
first := !d.waiting
|
||||
d.waiting = true
|
||||
d.mu.Unlock()
|
||||
if first {
|
||||
d.logf("the seat git did not answer the read of %d contracts modules' trunk definitions %s: %v; the indexer judges from the definitions it kept after %s, and reads them again once the seat answers", n, when, err, d.bound())
|
||||
if d.Waits != nil {
|
||||
d.Waits.WaitFor(ctx, GitDefinitionsView)
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
if err != nil {
|
||||
d.logf("reading the seat definitions of %d contracts modules' trunks %s: %v; each is read again at its next merged message or by indexer_seat_definitions_again", n, when, err)
|
||||
} else {
|
||||
d.logf("read the seat definitions of %d contracts modules' trunks %s", n, when)
|
||||
}
|
||||
d.mu.Lock()
|
||||
was := d.waiting
|
||||
d.waiting = false
|
||||
select {
|
||||
case <-d.answered:
|
||||
default:
|
||||
close(d.answered)
|
||||
}
|
||||
again := d.fromKept
|
||||
d.fromKept = false
|
||||
d.mu.Unlock()
|
||||
if was && d.Waits != nil {
|
||||
d.Waits.Holds(GitDefinitionsView)
|
||||
}
|
||||
if again && d.Judge != nil {
|
||||
if err := d.Judge(ctx); err != nil {
|
||||
d.logf("judging again the manifests judged from kept definitions: %v; judged again at the next message heard", err)
|
||||
} else {
|
||||
d.logf("judged again every manifest judged from kept definitions, now that the seat git answers")
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (d *DefinitionsAtStart) bound() time.Duration {
|
||||
if d.Bound > 0 {
|
||||
return d.Bound
|
||||
}
|
||||
return GitWaitBound
|
||||
}
|
||||
|
||||
// Ready waits, while the seat git does not answer the read at start, until a read is answered, or Bound passed (no timer: a context's bound), and says whether the
|
||||
// judging that follows runs on the definitions the indexer kept; it marks it so, and the read that answers later judges
|
||||
// again (Read's answer).
|
||||
func (d *DefinitionsAtStart) Ready(ctx context.Context) (fromKept bool) {
|
||||
d.mu.Lock()
|
||||
d.init()
|
||||
answered, waiting := d.answered, d.waiting
|
||||
d.mu.Unlock()
|
||||
if !waiting {
|
||||
return false // read, or not read at all (the seat definitions' state not bound yet): nothing to wait for
|
||||
}
|
||||
wait, cancel := context.WithTimeout(ctx, d.bound())
|
||||
defer cancel()
|
||||
select {
|
||||
case <-answered:
|
||||
return false
|
||||
case <-wait.Done():
|
||||
}
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
select {
|
||||
case <-d.answered:
|
||||
return false
|
||||
default:
|
||||
}
|
||||
d.fromKept = true
|
||||
d.logf("the seat git has not answered within %s: the indexer judges from the definitions it kept", d.bound())
|
||||
return true
|
||||
}
|
||||
|
||||
// Heard is a sign that the forge runs (a message the forge bridge relays): while the indexer waits for the seat git, it
|
||||
// reads the trunk definitions again.
|
||||
func (d *DefinitionsAtStart) Heard(ctx context.Context) {
|
||||
d.mu.Lock()
|
||||
waiting := d.waiting
|
||||
d.mu.Unlock()
|
||||
if waiting {
|
||||
d.read(ctx, "once the forge was heard")
|
||||
}
|
||||
}
|
||||
|
||||
// Waiting says the indexer waits for the seat git.
|
||||
func (d *DefinitionsAtStart) Waiting() bool {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.waiting
|
||||
}
|
||||
Reference in New Issue
Block a user