Merge pull request 'compose-candidate reads each manifest once and answers within its bound, from each ask's arrival' (#200) from fix/1126-compose-wait into main
This commit was merged in pull request #200.
This commit is contained in:
@@ -363,7 +363,8 @@ func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[s
|
|||||||
}
|
}
|
||||||
|
|
||||||
// composeCandidateFrom answers the verb compose-candidate (novox/hq ADR 0593 §7): the check's candidate composed on this
|
// composeCandidateFrom answers the verb compose-candidate (novox/hq ADR 0593 §7): the check's candidate composed on this
|
||||||
// node in shadow by the run's shadow, applying and writing nothing, within delivery.ComposeAnswerBound. The answer is
|
// node in shadow by the run's shadow, applying and writing nothing, within the shadow's own bound (shorter than
|
||||||
|
// delivery.ComposeAnswerBound, selfcompose candidateWithin). The answer is
|
||||||
// the seat's shape (delivery.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet
|
// the seat's shape (delivery.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet
|
||||||
// answers unready, which the checker takes as no answer.
|
// answers unready, which the checker takes as no answer.
|
||||||
func composeCandidateFrom(node string, shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) {
|
func composeCandidateFrom(node string, shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) {
|
||||||
|
|||||||
@@ -134,10 +134,28 @@ func runCase(r Request, c Case, depth string) (CaseVerdict, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return CaseVerdict{}, fmt.Errorf("composer: the case %q: %w", c.Name, err)
|
return CaseVerdict{}, fmt.Errorf("composer: the case %q: %w", c.Name, err)
|
||||||
}
|
}
|
||||||
|
// The module the mesh provides itself (the private network) has no repository the indexer reads, so no case carries
|
||||||
|
// it: its manifest comes with mesh-sdk and always wins, as the bus reader has it (frombus readBase). Without it a node
|
||||||
|
// it is assigned to composed nothing (issue 1126, check of mesh-host #200: "nothing provides mesh-wireguard").
|
||||||
|
for _, name := range delivery.ProvidedModules() {
|
||||||
|
m, _, err := delivery.ProvidedManifest(name)
|
||||||
|
if err != nil {
|
||||||
|
return CaseVerdict{}, fmt.Errorf("composer: %w", err)
|
||||||
|
}
|
||||||
|
shelf[name] = m
|
||||||
|
}
|
||||||
|
// A module the input carries as its node-engine took it in from its build (SentInput) is composed so: the bus
|
||||||
|
// reader composes it from its build, never from the indexer's manifest a case gives.
|
||||||
|
for name, m := range base.Catalogue {
|
||||||
|
shelf[name] = m
|
||||||
|
}
|
||||||
in.Catalogue = shelf
|
in.Catalogue = shelf
|
||||||
in.Nodes = slices.Clone(in.Nodes)
|
in.Nodes = slices.Clone(in.Nodes)
|
||||||
in.Holdings = slices.Clone(in.Holdings)
|
in.Holdings = slices.Clone(in.Holdings)
|
||||||
for i := range in.Nodes {
|
for i := range in.Nodes {
|
||||||
|
if len(c.Kept[in.Nodes[i].Name]) == 0 {
|
||||||
|
continue // what the input keeps of this node, as its engine took it in, stands
|
||||||
|
}
|
||||||
in.Nodes[i].Kept = nil
|
in.Nodes[i].Kept = nil
|
||||||
if kept := c.Kept[in.Nodes[i].Name]; len(kept) > 0 {
|
if kept := c.Kept[in.Nodes[i].Name]; len(kept) > 0 {
|
||||||
k, err := parse(kept, nil, nil)
|
k, err := parse(kept, nil, nil)
|
||||||
@@ -234,3 +252,38 @@ const Package = "github.com/novox/mesh-host/cmd/mesh-compose"
|
|||||||
|
|
||||||
// Decode reads a request (sdk.Decode); it refuses one that is not JSON or of another version.
|
// Decode reads a request (sdk.Decode); it refuses one that is not JSON or of another version.
|
||||||
func Decode(data []byte) (Request, error) { return sdk.Decode(data) }
|
func Decode(data []byte) (Request, error) { return sdk.Decode(data) }
|
||||||
|
|
||||||
|
// SentInput is the composition input a node-engine answers for the sandbox (delivery.ComposeCandidateAsk.Input), from the
|
||||||
|
// input it composed node from: the one shape both read (issue 1126, checks of mesh-host #200). Every other node's modules
|
||||||
|
// are left to the cases (the indexer's manifests, as the bus reader reads them too); node's own placed modules stay as the
|
||||||
|
// bus reader took them in from their builds (their artifacts, the texts of their templates archive, what is handed to the
|
||||||
|
// provisioner), with what node keeps at another build, which no case can give: the sandbox has no build or archive to
|
||||||
|
// read. Without them the sandbox composed another node than the engine (restic's templates and artifacts unread, postgres's
|
||||||
|
// secret reader handed and the module not).
|
||||||
|
func SentInput(in compose.Input, node string) compose.Input {
|
||||||
|
own := map[string]bool{}
|
||||||
|
for _, n := range in.Nodes {
|
||||||
|
if n.Name == node {
|
||||||
|
for _, m := range n.Assigned {
|
||||||
|
own[m] = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
taken := in.Catalogue
|
||||||
|
in.Catalogue = nil
|
||||||
|
for name, m := range taken {
|
||||||
|
if own[name] {
|
||||||
|
if in.Catalogue == nil {
|
||||||
|
in.Catalogue = map[string]catalogue.Manifest{}
|
||||||
|
}
|
||||||
|
in.Catalogue[name] = m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
in.Nodes = slices.Clone(in.Nodes)
|
||||||
|
for i := range in.Nodes {
|
||||||
|
if in.Nodes[i].Name != node {
|
||||||
|
in.Nodes[i].Kept = nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return in
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,14 +1,18 @@
|
|||||||
package composer
|
package composer
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"os"
|
"os"
|
||||||
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
"github.com/novox/mesh-host/internal/compose"
|
"github.com/novox/mesh-host/internal/compose"
|
||||||
|
"github.com/novox/mesh-host/internal/compose/frombus"
|
||||||
|
"github.com/novox/mesh-host/internal/compose/frombus/frombustest"
|
||||||
)
|
)
|
||||||
|
|
||||||
const demo = `{"module": "demo", "version": "1", "claims": [{"name": "node-files"}]}`
|
const demo = `{"module": "demo", "version": "1", "claims": [{"name": "node-files"}]}`
|
||||||
@@ -60,6 +64,50 @@ func TestARequestItCannotRunIsRefused(t *testing.T) {
|
|||||||
if _, err := Run(placed); err == nil || !strings.Contains(err.Error(), "holds no node for") {
|
if _, err := Run(placed); err == nil || !strings.Contains(err.Error(), "holds no node for") {
|
||||||
t.Errorf("a case placing a node the input lacks: %v", err)
|
t.Errorf("a case placing a node the input lacks: %v", err)
|
||||||
}
|
}
|
||||||
|
// Issue 1126 (check of mesh-host #200): the module the mesh provides itself (the private network) comes from mesh-sdk,
|
||||||
|
// a second source beside the indexer's manifests the cases carry, and the input carries no catalogue: a node it is
|
||||||
|
// assigned to still composes, and what it provides is provided.
|
||||||
|
wg := delivery.ProvidedModules()[0]
|
||||||
|
provided := withInput(t, Request{V: Version, Machines: []string{"a"}, Depth: DepthResolve,
|
||||||
|
Cases: []Case{{Name: "change", Manifests: map[string]Manifest{"demo": {Raw: json.RawMessage(demo)}}}}},
|
||||||
|
compose.Input{Nodes: []compose.Node{{Name: "a", Assigned: []string{wg, "demo"}}}})
|
||||||
|
if v, err := Run(provided); err != nil || !v.Cases[0].Machines["a"].Composes {
|
||||||
|
t.Errorf("a node assigned the mesh's own module: %+v, %v", v, err)
|
||||||
|
}
|
||||||
|
// Issue 1126 (checks of mesh-host #200, fix-a-bug-met-twice): the sandbox composes one node from the same inputs as
|
||||||
|
// the bus reader, through one path. A mesh whose restic is built (its files from an archive) and handed to the
|
||||||
|
// provisioner composes n1 through frombus, and the request a node-engine answers for it (the input as the engine
|
||||||
|
// sends it, the indexer's manifests in the case) declares the same paths for n1.
|
||||||
|
t.Run("the sandbox composes a node as the bus reader does", func(t *testing.T) {
|
||||||
|
b, err := frombustest.Provisioned()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
ctx, scope := context.Background(), frombus.Scope{Node: frombustest.Node}
|
||||||
|
read, err := frombus.Input(ctx, b, frombus.Mesh{}, scope)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
d, problems, err := compose.Compose(read, frombustest.Node)
|
||||||
|
if err != nil || len(problems) > 0 {
|
||||||
|
t.Fatalf("the bus reader's input composes n1: %v %v", err, problems)
|
||||||
|
}
|
||||||
|
want := pathsOf(d)
|
||||||
|
sent := SentInput(read, frombustest.Node)
|
||||||
|
cases := Case{Name: "change", Manifests: map[string]Manifest{}}
|
||||||
|
for name, m := range b.Module {
|
||||||
|
cases.Manifests[name] = Manifest{Raw: b.Manifests[m.Manifest]}
|
||||||
|
}
|
||||||
|
r := withInput(t, Request{V: Version, Machines: []string{frombustest.Node}, Declarations: true, Cases: []Case{cases}}, sent)
|
||||||
|
v, err := Run(r)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
got := v.Cases[0].Machines[frombustest.Node]
|
||||||
|
if !got.Composes || !reflect.DeepEqual(got.Declared, want) {
|
||||||
|
t.Errorf("the sandbox composed n1 %+v; the bus reader declared %+v", got, want)
|
||||||
|
}
|
||||||
|
})
|
||||||
v, err := Run(Request{V: Version, Cases: []Case{{Name: "only-read", Manifests: map[string]Manifest{"demo": {Raw: json.RawMessage(demo)}}}}})
|
v, err := Run(Request{V: Version, Cases: []Case{{Name: "only-read", Manifests: map[string]Manifest{"demo": {Raw: json.RawMessage(demo)}}}}})
|
||||||
if err != nil || len(v.Cases) != 1 || !v.Cases[0].Manifests["demo"].Accepted || v.Cases[0].Machines != nil || v.SDK == "" {
|
if err != nil || len(v.Cases) != 1 || !v.Cases[0].Manifests["demo"].Accepted || v.Cases[0].Machines != nil || v.SDK == "" {
|
||||||
t.Fatalf("a request with no machines: %+v, %v", v, err)
|
t.Fatalf("a request with no machines: %+v, %v", v, err)
|
||||||
|
|||||||
@@ -116,11 +116,10 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
apart := make(chan struct{}, apartAtOnce)
|
||||||
for _, v := range catalogue.EngineVerbs() {
|
for _, v := range catalogue.EngineVerbs() {
|
||||||
verb := v.Name
|
verb := v.Name
|
||||||
answer := micro.HandlerFunc(func(r micro.Request) {
|
answer := EngineAnswer(handle, verb, l.node, apart)
|
||||||
_ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data()))
|
|
||||||
})
|
|
||||||
meta := EngineEndpointMetadata(v, l.node)
|
meta := EngineEndpointMetadata(v, l.node)
|
||||||
_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
|
_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
|
||||||
micro.WithEndpointMetadata(meta))
|
micro.WithEndpointMetadata(meta))
|
||||||
@@ -130,6 +129,40 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
|
|||||||
l.engine = svc
|
l.engine = svc
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// answeredApart are the verbs answered off the subscription that delivered them: each endpoint's subscription hands its
|
||||||
|
// asks over one at a time, so a second ask of a verb that takes long waited for the first, and its bound counted from
|
||||||
|
// when the first ended (issue 1126, review of 2026-11-03: two checks asking compose-candidate of one node at once, the
|
||||||
|
// second answered at about 100 s, past the bus's one-minute reply window). Each is answered in a goroutine of its own,
|
||||||
|
// its bound counted from its arrival.
|
||||||
|
var answeredApart = map[string]bool{ActComposeCandidate: true}
|
||||||
|
|
||||||
|
// apartAtOnce bounds the asks answered apart at once on this node; one more is refused in words at once.
|
||||||
|
const apartAtOnce = 4
|
||||||
|
|
||||||
|
// EngineAnswer is the handler of one verb: answered on the subscription's goroutine, or, for a verb answered apart, in
|
||||||
|
// its own while fewer than apart's capacity run, and refused at once past it.
|
||||||
|
func EngineAnswer(handle EngineFunc, verb, node string, apart chan struct{}) micro.HandlerFunc {
|
||||||
|
return func(r micro.Request) {
|
||||||
|
if !answeredApart[verb] {
|
||||||
|
_ = r.Respond(AnswerEngineCall(handle, verb, node, r.Subject(), r.Data()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case apart <- struct{}{}:
|
||||||
|
default:
|
||||||
|
raw, _ := json.Marshal(map[string]any{"node": node, "error": fmt.Sprintf(
|
||||||
|
"this node answers %d asks of %s at once now: ask again when one is answered", cap(apart), verb)})
|
||||||
|
_ = r.Respond(raw)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
subject, data := r.Subject(), r.Data()
|
||||||
|
go func() {
|
||||||
|
defer func() { <-apart }()
|
||||||
|
_ = r.Respond(AnswerEngineCall(handle, verb, node, subject, data))
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's.
|
// EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's.
|
||||||
const EngineModule = "mesh-host"
|
const EngineModule = "mesh-host"
|
||||||
|
|
||||||
|
|||||||
@@ -7,11 +7,13 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
"github.com/nats-io/nats.go"
|
"github.com/nats-io/nats.go"
|
||||||
|
"github.com/nats-io/nats.go/micro"
|
||||||
)
|
)
|
||||||
|
|
||||||
// The hand-over's ask and answer hold these field names; the controller's side holds the same list
|
// The hand-over's ask and answer hold these field names; the controller's side holds the same list
|
||||||
@@ -265,7 +267,62 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
|
|||||||
t.Errorf("the judgements are recorded as %+v", got)
|
t.Errorf("the judgements are recorded as %+v", got)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
// Issue 1126 (review of 2026-11-03): two asks of compose-candidate at once, handed over one at a time as an
|
||||||
|
// endpoint's subscription hands them, are both answered while the other runs: neither waits for the other, so each
|
||||||
|
// bound counts from its own arrival. A verb answered on the subscription answers in order, as before.
|
||||||
|
t.Run("two asks of compose-candidate at once are answered side by side", func(t *testing.T) {
|
||||||
|
var mu sync.Mutex
|
||||||
|
arrived, both := 0, make(chan struct{})
|
||||||
|
handle := func(c EngineCall) (any, error) {
|
||||||
|
mu.Lock()
|
||||||
|
if arrived++; arrived == 2 {
|
||||||
|
close(both)
|
||||||
}
|
}
|
||||||
|
mu.Unlock()
|
||||||
|
select {
|
||||||
|
case <-both:
|
||||||
|
return "together", nil
|
||||||
|
case <-time.After(2 * time.Second): // only when the asks are answered one after the other
|
||||||
|
return "alone", nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
answered := make(chan string, 2)
|
||||||
|
answer := EngineAnswer(handle, ActComposeCandidate, "n1", make(chan struct{}, 4))
|
||||||
|
for range 2 {
|
||||||
|
answer(fakeRequest{subject: EngineToolSubject(ActComposeCandidate, "n1"), answered: answered})
|
||||||
|
}
|
||||||
|
for range 2 {
|
||||||
|
if got := <-answered; !strings.Contains(got, `"result":"together"`) {
|
||||||
|
t.Errorf("an ask answered %s; want both answered side by side", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
full := make(chan struct{}, 1)
|
||||||
|
full <- struct{}{}
|
||||||
|
EngineAnswer(handle, ActComposeCandidate, "n1", full)(fakeRequest{subject: "s", answered: answered})
|
||||||
|
if got := <-answered; !strings.Contains(got, "asks of compose-candidate at once") {
|
||||||
|
t.Errorf("an ask past the bound answered %s", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// fakeRequest is a seat call as the subscription hands it over, its answer sent to answered.
|
||||||
|
type fakeRequest struct {
|
||||||
|
subject string
|
||||||
|
answered chan string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r fakeRequest) Respond(b []byte, _ ...micro.RespondOpt) error {
|
||||||
|
r.answered <- string(b)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
func (r fakeRequest) RespondJSON(any, ...micro.RespondOpt) error { return nil }
|
||||||
|
func (r fakeRequest) Error(string, string, []byte, ...micro.RespondOpt) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
func (r fakeRequest) Data() []byte { return []byte(`{}`) }
|
||||||
|
func (r fakeRequest) Headers() micro.Headers { return nil }
|
||||||
|
func (r fakeRequest) Subject() string { return r.subject }
|
||||||
|
func (r fakeRequest) Reply() string { return "_INBOX.x" }
|
||||||
|
|
||||||
// **Against a real server**: a signed request on the node's own subject is answered; one on another node's is
|
// **Against a real server**: a signed request on the node's own subject is answered; one on another node's is
|
||||||
// not heard; a message with no reply is answered by nobody.
|
// not heard; a message with no reply is answered by nobody.
|
||||||
|
|||||||
@@ -6,13 +6,16 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"maps"
|
"maps"
|
||||||
|
"slices"
|
||||||
"sort"
|
"sort"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
|
||||||
"github.com/novox/mesh-host/internal/agent"
|
"github.com/novox/mesh-host/internal/agent"
|
||||||
"github.com/novox/mesh-host/internal/compose/frombus"
|
"github.com/novox/mesh-host/internal/compose/frombus"
|
||||||
|
"github.com/novox/mesh-host/internal/composer"
|
||||||
)
|
)
|
||||||
|
|
||||||
// **A check's candidate, composed in shadow on a call** (novox/hq ADR 0593 §7): the node seat's verb compose-candidate.
|
// **A check's candidate, composed in shadow on a call** (novox/hq ADR 0593 §7): the node seat's verb compose-candidate.
|
||||||
@@ -28,17 +31,102 @@ import (
|
|||||||
// errNotReady is a shadow that has not read every input yet.
|
// errNotReady is a shadow that has not read every input yet.
|
||||||
var errNotReady = errors.New("this node's shadow has not read every input yet")
|
var errNotReady = errors.New("this node's shadow has not read every input yet")
|
||||||
|
|
||||||
// ComposeCandidate is the node's answer to ask, within delivery.ComposeAnswerBound (or ctx's deadline, if sooner).
|
// candidateWithin is the node's own bound on an answer to compose-candidate: shorter than the checker's
|
||||||
|
// (delivery.ComposeAnswerBound), because the bus lets the engine answer an ask only within a minute of it (the grants
|
||||||
|
// part's allow_responses ttl, mesh-grants ResponseTTL): an answer at the checker's 60 s bound was refused by the bus
|
||||||
|
// (issue 1126, 2026-11-03 01:17:04 on shanks, "Permissions Violation for Publish to _INBOX..."), so the checker heard
|
||||||
|
// nothing at all. A node not done by then answers unready, naming what it waits on.
|
||||||
|
var candidateWithin = delivery.ComposeAnswerBound - 10*time.Second
|
||||||
|
|
||||||
|
// maxSentInput is the most composition input an answer carries: the bus carries 1 MiB in one message (its max_payload),
|
||||||
|
// and the answer's own fields and envelope come beside it.
|
||||||
|
const maxSentInput = 900 << 10
|
||||||
|
|
||||||
|
// warmAtOnce is how many of the indexer's manifests the shadow reads at once before it composes a candidate.
|
||||||
|
const warmAtOnce = 8
|
||||||
|
|
||||||
|
// candidateRun is what one ask's compositions wait on: each scratch composer it made, the last the one composing now.
|
||||||
|
type candidateRun struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
parts []*Composer
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *candidateRun) add(c *Composer) *Composer {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
r.parts = append(r.parts, c)
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
|
||||||
|
// said is the run's waits in words: how many reads it waited on the bus or the vault for, and the longest; and, while
|
||||||
|
// one is still running (now), what it waits on and since when.
|
||||||
|
func (r *candidateRun) said(now time.Time) (waits string, on string) {
|
||||||
|
r.mu.Lock()
|
||||||
|
parts := slices.Clone(r.parts)
|
||||||
|
r.mu.Unlock()
|
||||||
|
n, longest, longestOn := 0, time.Duration(0), ""
|
||||||
|
for _, c := range parts {
|
||||||
|
c.doingMu.Lock()
|
||||||
|
n += c.waits.n
|
||||||
|
if c.waits.longest > longest {
|
||||||
|
longest, longestOn = c.waits.longest, c.waits.longestOn
|
||||||
|
}
|
||||||
|
if c.doing != "" {
|
||||||
|
on = fmt.Sprintf("%s (for %s)", c.doing, now.Sub(c.doingSince).Round(time.Millisecond))
|
||||||
|
} else if on == "" && c.waits.lastOn != "" {
|
||||||
|
on = c.waits.lastOn + " (the last read)"
|
||||||
|
}
|
||||||
|
c.doingMu.Unlock()
|
||||||
|
}
|
||||||
|
waits = fmt.Sprintf("%d read(s) waited on", n)
|
||||||
|
if longestOn != "" {
|
||||||
|
waits += fmt.Sprintf(", the longest %s on %s", longest.Round(time.Millisecond), longestOn)
|
||||||
|
}
|
||||||
|
if on == "" {
|
||||||
|
on = "nothing it reads (composing)"
|
||||||
|
}
|
||||||
|
return waits, on
|
||||||
|
}
|
||||||
|
|
||||||
|
// ComposeCandidate is the node's answer to ask, within candidateWithin (or ctx's deadline, if sooner). Each answer is
|
||||||
|
// said in one line, with what it waited on and for how long (issue 1126: ace, g14 and shanks answered nothing for a
|
||||||
|
// minute while nothing said why). An answer not composed within the bound is unready, naming what the node waits on.
|
||||||
func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer {
|
func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer {
|
||||||
began := time.Now()
|
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
c := s.c
|
c := s.c
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
|
began := c.now()
|
||||||
|
ctx, cancel := context.WithTimeout(ctx, candidateWithin)
|
||||||
|
defer cancel()
|
||||||
|
run := &candidateRun{}
|
||||||
|
done := make(chan delivery.ComposeCandidateAnswer, 1)
|
||||||
|
go func() { done <- s.composeCandidate(ctx, c, ask, began, run) }()
|
||||||
|
var a delivery.ComposeCandidateAnswer
|
||||||
|
select {
|
||||||
|
case a = <-done:
|
||||||
|
case <-ctx.Done():
|
||||||
|
_, on := run.said(c.now())
|
||||||
|
a = delivery.ComposeCandidateAnswer{Node: c.env.Node, Candidate: ask.Candidate, Verdict: delivery.ComposeCandidateUnready,
|
||||||
|
Refused: fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on),
|
||||||
|
Took: c.now().Sub(began).Milliseconds()}
|
||||||
|
}
|
||||||
|
waits, _ := run.said(c.now())
|
||||||
|
words := ""
|
||||||
|
if a.Refused != "" {
|
||||||
|
words = ": " + a.Refused
|
||||||
|
}
|
||||||
|
c.env.Say(fmt.Sprintf("compose-candidate %s answered %s in %s (%s)%s", ask.Candidate, a.Verdict,
|
||||||
|
(time.Duration(a.Took) * time.Millisecond).String(), waits, words))
|
||||||
|
return a
|
||||||
|
}
|
||||||
|
|
||||||
|
// composeCandidate composes ask on scratch copies of c, each added to run as it starts.
|
||||||
|
func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery.ComposeCandidateAsk, began time.Time, run *candidateRun) delivery.ComposeCandidateAnswer {
|
||||||
node := c.env.Node
|
node := c.env.Node
|
||||||
a := delivery.ComposeCandidateAnswer{Node: node, Candidate: ask.Candidate}
|
a := delivery.ComposeCandidateAnswer{Node: node, Candidate: ask.Candidate}
|
||||||
refuse := func(why string) delivery.ComposeCandidateAnswer {
|
refuse := func(why string) delivery.ComposeCandidateAnswer {
|
||||||
a.Verdict, a.Refused, a.Modules, a.Own = delivery.ComposeCandidateRefused, why, nil, delivery.ModuleComposed{}
|
a.Verdict, a.Refused, a.Modules, a.Own = delivery.ComposeCandidateRefused, why, nil, delivery.ModuleComposed{}
|
||||||
a.Took = time.Since(began).Milliseconds()
|
a.Took = c.now().Sub(began).Milliseconds()
|
||||||
return a
|
return a
|
||||||
}
|
}
|
||||||
// A node that cannot compose now (its shadow not ready, a read the bus did not answer, the bound passed) answers
|
// A node that cannot compose now (its shadow not ready, a read the bus did not answer, the bound passed) answers
|
||||||
@@ -58,12 +146,10 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return refuse(err.Error())
|
return refuse(err.Error())
|
||||||
}
|
}
|
||||||
ctx, cancel := context.WithTimeout(ctx, delivery.ComposeAnswerBound)
|
|
||||||
defer cancel()
|
|
||||||
if !s.ready.Load() {
|
if !s.ready.Load() {
|
||||||
return unready(errNotReady.Error())
|
return unready(errNotReady.Error())
|
||||||
}
|
}
|
||||||
base := c.scratch()
|
base := run.add(c.scratch())
|
||||||
raw, err := base.get(ctx, delivery.CandidatesBucket, ckey)
|
raw, err := base.get(ctx, delivery.CandidatesBucket, ckey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return unready(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err))
|
return unready(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err))
|
||||||
@@ -80,17 +166,21 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
|
|||||||
// not asked about. No answer (ADR 0593 §9), never a verdict on the wrong heads.
|
// not asked about. No answer (ADR 0593 §9), never a verdict on the wrong heads.
|
||||||
return unready(fmt.Sprintf("the indexer's candidate of the group %s is at heads %s, not %s: the check asked is replaced", group, cand.HeadsHash, heads))
|
return unready(fmt.Sprintf("the indexer's candidate of the group %s is at heads %s, not %s: the check asked is replaced", group, cand.HeadsHash, heads))
|
||||||
}
|
}
|
||||||
laid := c.scratch()
|
// Every module's manifest is read before either composition, several at once, into the shadow's own: a composition
|
||||||
touched, err := laid.layCandidate(ctx, cand)
|
// reads every manifest in the mesh one by one, which off the control-node took about 35 s each (issue 1126).
|
||||||
if err != nil {
|
base.warm(ctx, c, cand)
|
||||||
return unready(err.Error())
|
|
||||||
}
|
|
||||||
|
|
||||||
var running []delivery.Composed
|
var running []delivery.Composed
|
||||||
if c.env.Running != nil {
|
if c.env.Running != nil {
|
||||||
running = c.env.Running()
|
running = c.env.Running()
|
||||||
}
|
}
|
||||||
baseComps, baseHeld, baseErr := base.composeShadow(ctx, running, nil, nil)
|
baseComps, baseHeld, baseErr := base.composeShadow(ctx, running, nil, nil)
|
||||||
|
// The candidate laid over a copy taken after the composition without it, so what that read is not read again.
|
||||||
|
laid := run.add(c.scratch())
|
||||||
|
touched, err := laid.layCandidate(ctx, cand)
|
||||||
|
if err != nil {
|
||||||
|
return unready(err.Error())
|
||||||
|
}
|
||||||
laidComps, laidHeld, laidErr := laid.composeShadow(ctx, running, touched, func(scope frombus.Scope) {
|
laidComps, laidHeld, laidErr := laid.composeShadow(ctx, running, touched, func(scope frombus.Scope) {
|
||||||
if !ask.Input {
|
if !ask.Input {
|
||||||
return
|
return
|
||||||
@@ -99,17 +189,21 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// The request's input carries no catalogue (composer.Request.Input): the cases give the manifests.
|
// As the sandbox reads it (composer.SentInput, the one shape): this node's own modules as taken in from their
|
||||||
in.Catalogue = nil
|
// builds, every other module's manifest left to the cases.
|
||||||
for i := range in.Nodes {
|
if body, err := json.Marshal(composer.SentInput(in, laid.env.Node)); err == nil {
|
||||||
in.Nodes[i].Kept = nil
|
|
||||||
}
|
|
||||||
if body, err := json.Marshal(in); err == nil {
|
|
||||||
a.Input = body
|
a.Input = body
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
if len(a.Input) > maxSentInput {
|
||||||
|
// Past what the bus carries in one answer, the checker would hear nothing: refused, saying so.
|
||||||
|
size := len(a.Input)
|
||||||
|
a.Input = nil
|
||||||
|
return refuse(fmt.Sprintf("the composition input this node composed from is %d bytes, more than the %d an answer on the bus carries", size, maxSentInput))
|
||||||
|
}
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
return unready(fmt.Sprintf("the bound of %s passed before the node composed the candidate", delivery.ComposeAnswerBound))
|
_, on := run.said(c.now())
|
||||||
|
return unready(fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on))
|
||||||
}
|
}
|
||||||
switch {
|
switch {
|
||||||
case laidErr != nil && BusFailed(laidErr):
|
case laidErr != nil && BusFailed(laidErr):
|
||||||
@@ -118,7 +212,7 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
|
|||||||
// The node composes none of its own without the candidate either: the candidate changes nothing of that.
|
// The node composes none of its own without the candidate either: the candidate changes nothing of that.
|
||||||
a.Verdict, a.Modules = delivery.ComposeCandidateComposed, []delivery.ModuleComposed{}
|
a.Verdict, a.Modules = delivery.ComposeCandidateComposed, []delivery.ModuleComposed{}
|
||||||
a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState, Words: laidErr.Error()}
|
a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState, Words: laidErr.Error()}
|
||||||
a.Took = time.Since(began).Milliseconds()
|
a.Took = c.now().Sub(began).Milliseconds()
|
||||||
return a
|
return a
|
||||||
case laidErr != nil:
|
case laidErr != nil:
|
||||||
return refuse(laidErr.Error())
|
return refuse(laidErr.Error())
|
||||||
@@ -193,7 +287,7 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
|
|||||||
if refused {
|
if refused {
|
||||||
a.Verdict = delivery.ComposeCandidateRefused
|
a.Verdict = delivery.ComposeCandidateRefused
|
||||||
}
|
}
|
||||||
a.Took = time.Since(began).Milliseconds()
|
a.Took = c.now().Sub(began).Milliseconds()
|
||||||
return a
|
return a
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -214,9 +308,63 @@ func (c *Composer) scratch() *Composer {
|
|||||||
out.ownHanded = maps.Clone(c.ownHanded)
|
out.ownHanded = maps.Clone(c.ownHanded)
|
||||||
out.certified = c.certified
|
out.certified = c.certified
|
||||||
out.placementKeys = c.placementKeys
|
out.placementKeys = c.placementKeys
|
||||||
|
out.origin = c
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// warm reads, several at once, each manifest the indexer's modules and the candidate name that the shadow (origin) does
|
||||||
|
// not hold yet, keeping each in origin and in c; one not read is left to the composition, which reads it and says why.
|
||||||
|
// Its reads are not noted as the bus failing: the composition that reads again notes its own.
|
||||||
|
func (c *Composer) warm(ctx context.Context, origin *Composer, cand delivery.Candidate) {
|
||||||
|
origin.mu.Lock()
|
||||||
|
want := map[string]bool{}
|
||||||
|
for _, m := range origin.modules {
|
||||||
|
if _, ok := origin.manifests[m.Manifest]; m.Manifest != "" && !ok {
|
||||||
|
want[m.Manifest] = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, pull := range cand.Pulls {
|
||||||
|
for _, cm := range pull.Fps {
|
||||||
|
if _, ok := origin.manifests[cm.Manifest]; cm.Manifest != "" && !ok {
|
||||||
|
want[cm.Manifest] = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
origin.mu.Unlock()
|
||||||
|
if len(want) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer c.waiting(fmt.Sprintf("%d of the indexer's manifests (%s)", len(want), delivery.ManifestsBucket))()
|
||||||
|
slots := make(chan struct{}, warmAtOnce)
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for digest := range want {
|
||||||
|
key, err := delivery.ManifestsKey(digest)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
select {
|
||||||
|
case slots <- struct{}{}:
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer func() { <-slots }()
|
||||||
|
value, err := c.env.Get(ctx, delivery.ManifestsBucket, key)
|
||||||
|
if err != nil || value == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
m, err := delivery.DecodeManifest(value)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
c.keepRead(func(k *Composer) { k.manifests[digest] = []byte(m.JSON) })
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
// layCandidate lays the candidate's modules over the indexer's: each at its fingerprint, with its manifest's digest and
|
// layCandidate lays the candidate's modules over the indexer's: each at its fingerprint, with its manifest's digest and
|
||||||
// the indexer's verdict on it (read from its verdicts, none read taken as accepted); a module the candidate removes goes.
|
// the indexer's verdict on it (read from its verdicts, none read taken as accepted); a module the candidate removes goes.
|
||||||
// It answers the modules laid.
|
// It answers the modules laid.
|
||||||
|
|||||||
@@ -132,4 +132,121 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
|
|||||||
if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {
|
if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {
|
||||||
t.Errorf("composing candidates wrote %v, marked %d have(s), handed %d answer(s); want nothing", state.written(), applied.Load(), handed.Load())
|
t.Errorf("composing candidates wrote %v, marked %d have(s), handed %d answer(s); want nothing", state.written(), applied.Load(), handed.Load())
|
||||||
}
|
}
|
||||||
|
// Issue 1126, 2026-11-03: off the control-node an ask of compose-candidate was not answered within a minute, and said
|
||||||
|
// nothing of why. Each of its two compositions read every manifest in the mesh one by one, on copies of the shadow
|
||||||
|
// that kept nothing, so every ask read them all twice again. Each manifest is now read once, before either
|
||||||
|
// composition, into the shadow, and a later ask reads none; an ask whose read does not answer is answered unready at
|
||||||
|
// its bound, naming what the node waits on; and every answer is said in one line with what it waited on.
|
||||||
|
t.Run("each manifest is read once, and the answer comes within its bound", func(t *testing.T) {
|
||||||
|
fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
|
||||||
|
fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
||||||
|
lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1)
|
||||||
|
other := strings.NewReplacer(`"module":"lemurs"`, `"module":"other"`, "lemurs.service", "other.service").Replace(lemurs)
|
||||||
|
digest, digest2, digestOther := delivery.ManifestDigest([]byte(lemurs)), delivery.ManifestDigest([]byte(lemurs2)),
|
||||||
|
delivery.ManifestDigest([]byte(other))
|
||||||
|
otherKey, _ := delivery.ManifestsKey(digestOther)
|
||||||
|
|
||||||
|
// shadow is a ready shadow of n1 over b: lemurs placed on it, other placed nowhere (its manifest read for the mesh's
|
||||||
|
// catalogue alone), and a candidate laying lemurs at a new build; its clock moves a second at each reading.
|
||||||
|
shadow := func(b *fakeBus) (*Shadow, string) {
|
||||||
|
for d, body := range map[string]string{digest: lemurs, digest2: lemurs2, digestOther: other} {
|
||||||
|
man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "x", JSON: body})
|
||||||
|
key, _ := delivery.ManifestsKey(d)
|
||||||
|
b.kv[delivery.ManifestsBucket+"/"+key] = man
|
||||||
|
}
|
||||||
|
head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee"}
|
||||||
|
heads, _ := delivery.HeadsHash([]delivery.Head{head})
|
||||||
|
pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"},
|
||||||
|
Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: digest2}}}}
|
||||||
|
raw, _ := delivery.EncodeCandidate(delivery.Candidate{Group: "fix/1126", HeadsHash: heads, Pulls: pulls,
|
||||||
|
FPSet: delivery.FingerprintSet(pulls)})
|
||||||
|
ckey, _ := delivery.CandidatesKey("fix/1126")
|
||||||
|
b.kv[delivery.CandidatesBucket+"/"+ckey] = raw
|
||||||
|
env := b.env()
|
||||||
|
var tick atomic.Int64
|
||||||
|
env.Now = func() time.Time { return time.Unix(0, 0).Add(time.Duration(tick.Add(1)) * time.Second) }
|
||||||
|
s := NewShadow(env, nil)
|
||||||
|
s.c.placements["n1"] = delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||||
|
Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
|
||||||
|
s.c.records["n1"] = delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}
|
||||||
|
s.c.modules["lemurs"] = delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}
|
||||||
|
s.c.modules["other"] = delivery.Module{FP2: fp, Manifest: digestOther, Repository: "novox/x", Commit: "c"}
|
||||||
|
s.ready.Store(true)
|
||||||
|
key, _ := delivery.CompositionKey("fix/1126", heads)
|
||||||
|
return s, key
|
||||||
|
}
|
||||||
|
readsOf := func(b *fakeBus, key string) (n int) {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
for _, r := range b.reads {
|
||||||
|
if r == [2]string{delivery.ManifestsBucket, key} {
|
||||||
|
n++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
saidLast := func(b *fakeBus) string {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
if len(b.said) == 0 {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return b.said[len(b.said)-1]
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
// stuck: the read of other's manifest does not answer until the ask's bound ends.
|
||||||
|
stuck bool
|
||||||
|
asks int
|
||||||
|
verdict string
|
||||||
|
reads int // of other's manifest, over every ask
|
||||||
|
words string // in the answer's words
|
||||||
|
said string // in the line said of the last answer
|
||||||
|
}{
|
||||||
|
{name: "one ask reads each manifest once", asks: 1, verdict: delivery.ComposeCandidateComposed, reads: 1,
|
||||||
|
said: "compose-candidate fix/1126."},
|
||||||
|
{name: "a later ask reads none again", asks: 2, verdict: delivery.ComposeCandidateComposed, reads: 1,
|
||||||
|
said: "read(s) waited on"},
|
||||||
|
{name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: delivery.ComposeCandidateUnready,
|
||||||
|
words: "waits on", said: "answered unready"},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
b := newFakeBus()
|
||||||
|
s, key := shadow(b)
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
if tc.stuck {
|
||||||
|
get := s.c.env.Get
|
||||||
|
s.c.env.Get = func(rctx context.Context, bucket, k string) ([]byte, error) {
|
||||||
|
if bucket == delivery.ManifestsBucket && k == otherKey {
|
||||||
|
b.mu.Lock()
|
||||||
|
b.reads = append(b.reads, [2]string{bucket, k})
|
||||||
|
b.mu.Unlock()
|
||||||
|
cancel() // the ask's bound ends while the read waits
|
||||||
|
<-rctx.Done()
|
||||||
|
return nil, rctx.Err()
|
||||||
|
}
|
||||||
|
return get(rctx, bucket, k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
var a delivery.ComposeCandidateAnswer
|
||||||
|
for range tc.asks {
|
||||||
|
a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key})
|
||||||
|
}
|
||||||
|
if a.Verdict != tc.verdict || !strings.Contains(a.Refused, tc.words) {
|
||||||
|
t.Errorf("answered %s %q; want %s with %q", a.Verdict, a.Refused, tc.verdict, tc.words)
|
||||||
|
}
|
||||||
|
if tc.stuck && !strings.Contains(a.Refused, "manifest") {
|
||||||
|
t.Errorf("answered %q; want the manifest it waits on named", a.Refused)
|
||||||
|
}
|
||||||
|
if got := readsOf(b, otherKey); tc.reads > 0 && got != tc.reads {
|
||||||
|
t.Errorf("other's manifest was read %d time(s); want %d", got, tc.reads)
|
||||||
|
}
|
||||||
|
if line := saidLast(b); !strings.Contains(line, tc.said) {
|
||||||
|
t.Errorf("said %q; want a line with %q", line, tc.said)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -97,6 +97,8 @@ type Env struct {
|
|||||||
// ComposedState is this node's composition state `composed-<node>` (novox/hq ADR 0593 §6), where every composition
|
// ComposedState is this node's composition state `composed-<node>` (novox/hq ADR 0593 §6), where every composition
|
||||||
// a pass makes is published for its readers (publishComposed). Nil keeps none.
|
// a pass makes is published for its readers (publishComposed). Nil keeps none.
|
||||||
ComposedState ComposedStore
|
ComposedState ComposedStore
|
||||||
|
// Now is the clock what the composer waits on is timed by; nil is time.Now.
|
||||||
|
Now func() time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
|
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
|
||||||
@@ -143,6 +145,12 @@ type Composer struct {
|
|||||||
// it is read while c.mu is held.
|
// it is read while c.mu is held.
|
||||||
doingMu sync.Mutex
|
doingMu sync.Mutex
|
||||||
doing string
|
doing string
|
||||||
|
doingSince time.Time
|
||||||
|
// waits is what the composer waited on, under doingMu: for compose-candidate's line (issue 1126).
|
||||||
|
waits waitLog
|
||||||
|
// origin is the composer a scratch copy was taken of (scratch): what is read once by its digest or key and never
|
||||||
|
// changes (a manifest, a templates archive, a done build) is kept there too, so the next copy reads it no more.
|
||||||
|
origin *Composer
|
||||||
caught bool
|
caught bool
|
||||||
rev uint64
|
rev uint64
|
||||||
// missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one:
|
// missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one:
|
||||||
@@ -1060,19 +1068,56 @@ func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Compos
|
|||||||
return delivery.KeepSeals(comp, kept, c.sameValue)
|
return delivery.KeepSeals(comp, kept, c.sameValue)
|
||||||
}
|
}
|
||||||
|
|
||||||
// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before.
|
// waitLog is what a composer waited on: how many waits ended, the longest and what it was, and the last.
|
||||||
|
type waitLog struct {
|
||||||
|
n int
|
||||||
|
longest time.Duration
|
||||||
|
longestOn string
|
||||||
|
lastOn string
|
||||||
|
}
|
||||||
|
|
||||||
|
// now is the composer's clock (Env.Now).
|
||||||
|
func (c *Composer) now() time.Time {
|
||||||
|
if c.env.Now != nil {
|
||||||
|
return c.env.Now()
|
||||||
|
}
|
||||||
|
return time.Now()
|
||||||
|
}
|
||||||
|
|
||||||
|
// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before
|
||||||
|
// and counts the wait, timed, in waits.
|
||||||
func (c *Composer) waiting(what string) func() {
|
func (c *Composer) waiting(what string) func() {
|
||||||
|
began := c.now()
|
||||||
c.doingMu.Lock()
|
c.doingMu.Lock()
|
||||||
before := c.doing
|
before, beforeSince := c.doing, c.doingSince
|
||||||
c.doing = what
|
c.doing, c.doingSince = what, began
|
||||||
c.doingMu.Unlock()
|
c.doingMu.Unlock()
|
||||||
return func() {
|
return func() {
|
||||||
|
took := c.now().Sub(began)
|
||||||
c.doingMu.Lock()
|
c.doingMu.Lock()
|
||||||
c.doing = before
|
c.doing, c.doingSince = before, beforeSince
|
||||||
|
c.waits.n++
|
||||||
|
c.waits.lastOn = what
|
||||||
|
if took > c.waits.longest || c.waits.longestOn == "" {
|
||||||
|
c.waits.longest, c.waits.longestOn = took, what
|
||||||
|
}
|
||||||
c.doingMu.Unlock()
|
c.doingMu.Unlock()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// keepRead keeps, in c and in the composer c is a scratch copy of, what set writes: a read that never changes for its
|
||||||
|
// digest or key. Each under its own lock, one after the other.
|
||||||
|
func (c *Composer) keepRead(set func(*Composer)) {
|
||||||
|
c.mu.Lock()
|
||||||
|
set(c)
|
||||||
|
c.mu.Unlock()
|
||||||
|
if o := c.origin; o != nil {
|
||||||
|
o.mu.Lock()
|
||||||
|
set(o)
|
||||||
|
o.mu.Unlock()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// waitingOn is what the composer waits on now, in words.
|
// waitingOn is what the composer waits on now, in words.
|
||||||
func (c *Composer) waitingOn() string {
|
func (c *Composer) waitingOn() string {
|
||||||
c.doingMu.Lock()
|
c.doingMu.Lock()
|
||||||
@@ -1303,9 +1348,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
r.c.mu.Lock()
|
r.c.keepRead(func(k *Composer) { k.manifests[digest] = []byte(m.JSON) })
|
||||||
r.c.manifests[digest] = []byte(m.JSON)
|
|
||||||
r.c.mu.Unlock()
|
|
||||||
return []byte(m.JSON), nil
|
return []byte(m.JSON), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1330,9 +1373,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder
|
|||||||
return nil, fmt.Errorf("the builder's %s: %w", key, err)
|
return nil, fmt.Errorf("the builder's %s: %w", key, err)
|
||||||
}
|
}
|
||||||
if b.State == delivery.BuildDone {
|
if b.State == delivery.BuildDone {
|
||||||
r.c.mu.Lock()
|
r.c.keepRead(func(k *Composer) { k.builds[key] = b })
|
||||||
r.c.builds[key] = b
|
|
||||||
r.c.mu.Unlock()
|
|
||||||
}
|
}
|
||||||
return &b, nil
|
return &b, nil
|
||||||
}
|
}
|
||||||
@@ -1363,9 +1404,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error)
|
|||||||
if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
|
if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
|
||||||
return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
|
return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
|
||||||
}
|
}
|
||||||
r.c.mu.Lock()
|
r.c.keepRead(func(k *Composer) { k.archives[digest] = body })
|
||||||
r.c.archives[digest] = body
|
|
||||||
r.c.mu.Unlock()
|
|
||||||
return body, nil
|
return body, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user