136 lines
6.0 KiB
Go
136 lines
6.0 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-controller/internal/chain"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// The chain on the bus (novox/hq ADR 0408, design 47 milestone 5): builds, rollouts and assignments are keys of the
|
|
// bus's key-value store, written by the controller that holds the lease, and the controller's store keeps none of them.
|
|
|
|
// chainHolds is whether this process may write the chain now: the lease's gate, the one every act passes.
|
|
func chainHolds(ctx context.Context) error {
|
|
_, err := theLease.epoch(ctx)
|
|
return err
|
|
}
|
|
|
|
// openTheChain opens the chain's buckets on the bus: the serving controller's connection, or one dialled for a command.
|
|
func openTheChain(ctx context.Context) (*chain.Chain, error) {
|
|
js, err := aBus()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Not closed: a borrowed connection closes nothing, and a dialled one lasts as long as the command.
|
|
return chain.OnTheBus(ctx, js.Conn(), chainHolds)
|
|
}
|
|
|
|
// attachTheChain gives the inventory of the serving controller the chain on its connection.
|
|
func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.Server) error {
|
|
bus, ok := server.Bus().(link.OverNATS)
|
|
if !ok {
|
|
return errors.New("the chain lives on the bus, and this control plane is not on it")
|
|
}
|
|
c, err := chain.OnTheBus(ctx, bus.Conn, chainHolds)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
inv.UseChain(c)
|
|
inv.RequireLayout()
|
|
// And the builds laid out under their modules, once, before they are read (issue 1121) — **before this returns**, so
|
|
// nothing the start goes on to read meets a chain not laid out yet: on 2026-10-18 the layout ran beside the start, a
|
|
// read of the start was refused, and the controller exited and was started again every five seconds. The layout of
|
|
// keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly.
|
|
if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil {
|
|
fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err)
|
|
go layBuildsOutSoon(ctx, inv)
|
|
}
|
|
// The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build
|
|
// (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes.
|
|
// And what delivery 1 recorded of what each machine was sent, copied onto the keys of the assignments (CopySentBuilds).
|
|
go copySentBuildsSoon(ctx, inv)
|
|
return nil
|
|
}
|
|
|
|
// kindSentBuildsNotCopied is the condition of the one-time copy of node.sent_builds onto the assignments' keys that has not
|
|
// succeeded: until it has, a module an assignment pulled in may be missing from what a machine is read to run.
|
|
const kindSentBuildsNotCopied = "sent-builds-not-copied"
|
|
|
|
// copySentBuildsSoon copies what the store recorded of what each machine was sent onto the keys of the assignments, as soon
|
|
// as this controller holds the lease (the chain refuses a write before), and tries again every half minute while it fails,
|
|
// each time it fails saying so as an urgent condition until it has worked. It is idempotent: CopySentBuilds overwrites no key
|
|
// that names a commit and empties each column it has copied.
|
|
func copySentBuildsSoon(ctx context.Context, inv *inventory.Inventory) {
|
|
failed := false
|
|
for {
|
|
_, err := inv.CopySentBuilds(ctx)
|
|
o := conditions.Observation{Scope: conditions.ScopeBuild, ID: kindSentBuildsNotCopied, Token: kindSentBuildsNotCopied,
|
|
Kind: kindSentBuildsNotCopied, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator,
|
|
Source: "chain", Summary: "what each machine was sent is not copied onto its assignments: "}
|
|
if err != nil {
|
|
fmt.Printf("chain: what the machines were sent is not copied onto the assignments: %v\n", err)
|
|
if keeper := conditionsFrom; keeper != nil && !failed {
|
|
o.Summary += err.Error()
|
|
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
|
|
failed = true
|
|
}
|
|
}
|
|
} else {
|
|
if keeper := conditionsFrom; keeper != nil && failed {
|
|
_, _ = keeper.Clear(ctx, o.Key(), "copied")
|
|
}
|
|
return
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(chainMoveRetry):
|
|
}
|
|
}
|
|
}
|
|
|
|
// chainMoveRetry is how long the copy waits before it tries again.
|
|
const chainMoveRetry = 30 * time.Second
|
|
|
|
// kindBuildsNotLaidOut is the condition of the layout of the builds under their modules that has not succeeded: until it
|
|
// has, no build is read, and a send that needs one is refused with the reason.
|
|
const kindBuildsNotLaidOut = "builds-not-laid-out"
|
|
|
|
// layBuildsOutSoon lays the builds out under their modules as soon as this controller holds the lease (the chain refuses a
|
|
// write before), and tries again every half minute while it fails, loudly: each failure is printed, and the first raises an
|
|
// urgent condition until it has worked. It is idempotent (LayBuildsOut).
|
|
func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) {
|
|
failed := false
|
|
for {
|
|
err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) })
|
|
o := conditions.Observation{Scope: conditions.ScopeBuild, ID: kindBuildsNotLaidOut, Token: kindBuildsNotLaidOut,
|
|
Kind: kindBuildsNotLaidOut, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator,
|
|
Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "}
|
|
if err != nil {
|
|
fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err)
|
|
if keeper := conditionsFrom; keeper != nil && !failed {
|
|
o.Summary += err.Error()
|
|
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
|
|
failed = true
|
|
}
|
|
}
|
|
} else {
|
|
if keeper := conditionsFrom; keeper != nil && failed {
|
|
_, _ = keeper.Clear(ctx, o.Key(), "laid out")
|
|
}
|
|
return
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(chainMoveRetry):
|
|
}
|
|
}
|
|
}
|