diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go index 28f2895..ab78558 100644 --- a/internal/selfcompose/data.go +++ b/internal/selfcompose/data.go @@ -50,7 +50,13 @@ func (c *Composer) publishData(ctx context.Context) { if !due || c.env.Data == nil { return } - items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, frombus.Scope{Node: c.env.Node}) + c.mu.Lock() + scope := c.dataScope + c.mu.Unlock() + if scope.Node == "" { + scope.Node = c.env.Node + } + items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope) if err != nil { c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " + delivery.EngineDataBucket + " holds stands: " + err.Error()) diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index e33cd97..a676da9 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -145,4 +145,29 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { } }) } + // The items are worked out from what the pass read: a done build is read over the bus once, not again for the data + // items after the pass (a module in a running build is read again, since it may land). + b := newFakeBus() + done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte(lemurs))}) + running, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: "building"}) + fp := delivery.SourcePrefix2 + strings.Repeat("b", 64) + doneKey, _ := delivery.BuilderBuildKey("web", fp) + runningKey, _ := delivery.BuilderBuildKey("pg", fp) + b.kv[delivery.BuilderBuildsBucket+"/"+doneKey], b.kv[delivery.BuilderBuildsBucket+"/"+runningKey] = done, running + r := reader{c: New(b.env())} + for range 3 { + if got, err := r.Build(context.Background(), "web", fp); err != nil || got.State != delivery.BuildDone { + t.Fatalf("the done build: %+v, %v", got, err) + } + if got, err := r.Build(context.Background(), "pg", fp); err != nil || got.State != "building" { + t.Fatalf("the running build: %+v, %v", got, err) + } + } + reads := map[string]int{} + for _, read := range b.reads { + reads[read[1]]++ + } + if reads[doneKey] != 1 || reads[runningKey] != 3 { + t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads) + } } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index cb8b49d..a8d7b4d 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -113,7 +113,10 @@ type Composer struct { values map[string]map[string]any // by module, this node's manifests map[string][]byte // by digest, read once archives map[string][]byte // templates archives by digest, read once - handed map[string]string // composed key to the digest handed + // builds are the builder's done builds by key, read once: a done build at a fingerprint never changes, so a pass and + // the data items after it (publishData) do not read each placed module's build again over the bus (ADR 0577). + builds map[string]delivery.BuilderBuild + handed map[string]string // composed key to the digest handed // sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again // is other bytes, so the one first had is kept while the generation stands, and the composition stays the same. sealedOwn map[string]ownSealed @@ -152,9 +155,10 @@ type Composer struct { // dataDue says a pass composed something new (or is the run's first), so the data items are kept again // (publishData); dataSaid is the last line said of them. Under mu. dataKept is what this node's keys of the state // data hold, read once per run, nil before; the loop's alone. - dataDue bool - dataSaid string - dataKept map[string][]byte + dataDue bool + dataSaid string + dataScope frombus.Scope + dataKept map[string][]byte } // Pass is how one pass went, as the node-engine's verb compose answers it. @@ -234,7 +238,7 @@ func New(env Env) *Composer { } return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, placementKeys: delivery.NewPlacementKeys(), records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{}, - manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}, + manifests: map[string][]byte{}, archives: map[string][]byte{}, builds: map[string]delivery.BuilderBuild{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}, asked: make(chan chan Pass, 8)} } @@ -791,6 +795,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { // Its own certificate, asked of the vault when a module here asks for one and the record holds none it trusts. c.certifyIfWanted(ctx, r, own) scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running} + c.mu.Lock() + c.dataScope = frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn} // the data items are worked out as composed + c.mu.Unlock() missed := c.missed.Load() compositions, _, held, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps) err = c.missedIn(missed, err) @@ -1198,6 +1205,12 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder if err != nil { return nil, err } + r.c.mu.Lock() + done, ok := r.c.builds[key] + r.c.mu.Unlock() + if ok { + return &done, nil + } defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")() raw, err := r.c.get(ctx, delivery.BuilderBuildsBucket, key) if err != nil || raw == nil { @@ -1207,6 +1220,11 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder if err != nil { return nil, fmt.Errorf("the builder's %s: %w", key, err) } + if b.State == delivery.BuildDone { + r.c.mu.Lock() + r.c.builds[key] = b + r.c.mu.Unlock() + } return &b, nil }