587 lines
24 KiB
Go
587 lines
24 KiB
Go
// Command mesh-delivery-assigner is delivery's assigner (novox/hq ADR 0429 §1 and §6, ADR 0446, ADR 0528, issue 1297): a
|
|
// process on its own bus credential that watches what the indexer, the builder, the controller, the node-engines and
|
|
// the watchman write, and writes what each node should run (`want.<node>.<module>`, by the six rules), the
|
|
// fingerprints it holds, every difference between a want and what the controller sends, and the composition check's
|
|
// verdict of each group the checker judges. In release 5 the wants are only compared: the controller still sends what
|
|
// each node runs (ADR 0429 §7). It holds no timer: every pass is started by a change it watched.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"slices"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/announce"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
|
"git.novox.be/novox/mesh-sdk/go/health"
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"git.novox.be/novox/mesh-sdk/go/provider"
|
|
"git.novox.be/novox/mesh-sdk/go/provisions"
|
|
|
|
"mesh-delivery-assigner/internal/assign"
|
|
)
|
|
|
|
func main() {
|
|
if os.Getenv(process.BrokerFileEnv) == "" {
|
|
// Started by the tool runner, with no credential of its own: the put's tools (ADR 0546), served through the runtime,
|
|
// since a process never serves (ADR 0436). The process the manifest runs names its credential.
|
|
if err := serveTools(); err != nil {
|
|
fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err)
|
|
os.Exit(1)
|
|
}
|
|
return
|
|
}
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
if err := run(ctx, os.Getenv, os.Stderr); err != nil {
|
|
fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test
|
|
// holds the two lists equal), so the controller's broker grants exactly these.
|
|
var readBuckets = []string{
|
|
buscompose.ResolvedBucket,
|
|
provider.AssignmentsBucket,
|
|
delivery.PartGrantsBucket,
|
|
buscompose.LeaseBucket,
|
|
delivery.ControllerComposedBucket,
|
|
delivery.AnswersBucket,
|
|
delivery.BuilderBuildsBucket,
|
|
delivery.CandidatesBucket,
|
|
delivery.ManifestsBucket,
|
|
delivery.ModulesBucket,
|
|
delivery.PresenceBucket,
|
|
delivery.HaveBucket,
|
|
delivery.StepsBucket(goBuildModule),
|
|
takeoverBucket,
|
|
provisions.ProvisionsBucket,
|
|
keptBucket,
|
|
}
|
|
|
|
// readLogs are the logs of other parts the assigner reads, each named in its manifest's reads: the builder's log, whose
|
|
// newest done trunk build of a module names the commit a reader of the manifests runs at (novox/hq ADR 0571 clause 1).
|
|
var readLogs = []string{delivery.BuildLogRead}
|
|
|
|
// takeoverBucket is the condition keeper's take-over key's bucket (novox/hq ADR 0564 §9): whether the assigner's said
|
|
// conditions go straight to the router or only as findings (issue 1952 WM6).
|
|
var takeoverBucket = health.Bucket(health.KeeperModule, health.TakeoverState)
|
|
|
|
// goBuildModule is the Go build module, which takes the assigner's compose asks on its seat's check stream and writes
|
|
// their steps in its steps state (novox/hq ADR 0482 clause 4).
|
|
const goBuildModule = "mesh-build-go"
|
|
|
|
// ownBuckets are the assigner's own states, as its manifest declares them: the seat mesh-placement's three among them
|
|
// (ADR 0567 §1), named after the assigner as every module state is.
|
|
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket,
|
|
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket,
|
|
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, delivery.SeatHoldersBucket}
|
|
|
|
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
|
|
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
|
|
var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket,
|
|
delivery.PlacementPulsesBucket}
|
|
|
|
// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
|
|
// (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view
|
|
// and the teller, its puts by whoever follows one (ADR 0546), and the seat mesh-placement's placement, proposals and
|
|
// candidates by the node agents, the grants part, mesh-config and the view (ADR 0567 §9), and its placement by
|
|
// the builder, the checker and the indexer, through mesh-sdk's toolchain gate (issue 774, ADR 0549 §3).
|
|
var keptStates = []announce.State{
|
|
{Name: delivery.PlacementState, Readable: true},
|
|
{Name: delivery.PlacementProposalsState, Readable: true},
|
|
{Name: delivery.PlacementCandidatesState, Readable: true},
|
|
{Name: delivery.PlacementPulsesState, Readable: true},
|
|
{Name: "asks", Readable: true},
|
|
{Name: "compositions", Readable: true},
|
|
{Name: "differences", Readable: true},
|
|
{Name: "puts", Readable: true},
|
|
{Name: "wants", Readable: true},
|
|
{Name: health.FindingsState, Readable: true},
|
|
{Name: delivery.AssignerPoliciesState, Readable: true},
|
|
{Name: delivery.SeatHoldersState, Readable: true},
|
|
}
|
|
|
|
// buckets is every bucket the assigner opened, by name.
|
|
type buckets map[string]jetstream.KeyValue
|
|
|
|
// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named with its grant (issue 1945),
|
|
// since a pass without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it
|
|
// does not exist yet, the credential is not granted it yet, the bus does not answer in time), the assigner writes wants
|
|
// without it and takes no put further until it starts again (ADR 0546).
|
|
func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) (buckets, error) {
|
|
out := buckets{}
|
|
// The controller's lease first (ADR 0549 §5, ADR 0561 §3): once it is handed away the seat's placement alone names
|
|
// where modules run, and the controller's assignments are no bucket the assigner needs. Not granted or not there, the
|
|
// controller holds it, as before.
|
|
away, why := false, ""
|
|
if kv, err := js.KeyValue(ctx, buscompose.LeaseBucket); err != nil {
|
|
logf("the bucket %s cannot be opened (%v): the controller is taken to hold its lease, so the assignments are read beside the placement", buscompose.LeaseBucket, err)
|
|
} else {
|
|
out[buscompose.LeaseBucket] = kv
|
|
e, gerr := kv.Get(ctx, buscompose.ControllerLeaseKey)
|
|
var raw []byte
|
|
if gerr == nil {
|
|
raw = e.Value()
|
|
}
|
|
if errors.Is(gerr, jetstream.ErrKeyNotFound) {
|
|
gerr = nil
|
|
}
|
|
away, why = buscompose.LeaseHandedAway(raw, raw != nil, gerr)
|
|
}
|
|
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
|
if name == buscompose.LeaseBucket {
|
|
continue
|
|
}
|
|
kv, err := js.KeyValue(ctx, name)
|
|
if name == provider.AssignmentsBucket && err != nil && away {
|
|
logf("the bucket %s cannot be opened (%v), and none is needed: %s, so the seat's placement alone is read", name, err, why)
|
|
continue
|
|
}
|
|
if name == delivery.AssignerPutsBucket && err != nil {
|
|
logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
|
|
continue
|
|
}
|
|
if slices.Contains(seatBuckets, name) && err != nil {
|
|
// The seat's states come in the same delivery as this assigner (ADR 0567 §1): without placement the controller's
|
|
// assignments alone name where modules run, as before any placement was written (ADR 0549 §5).
|
|
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
|
continue
|
|
}
|
|
if name == delivery.SeatHoldersBucket && err != nil {
|
|
// The seat's state, new with ADR 0571 §4: bound once the bus says it was made; until then nobody is told who
|
|
// holds which seat here, and the controller's seats still answer.
|
|
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
|
continue
|
|
}
|
|
if name == keptBucket && err != nil {
|
|
// mesh-config's held values (ADR 0565 §5): without them a composition lays none, said in the journal.
|
|
logf("the bucket %s cannot be opened (%v): no held value is laid on a composition until it is bound", name, err)
|
|
continue
|
|
}
|
|
if name == provisions.ProvisionsBucket && err != nil {
|
|
// The seat mesh-provisioner's state, with its own delivery (ADR 0570 §4): bound once the bus says it was made;
|
|
// until then no want waits for a provision, as before.
|
|
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
|
continue
|
|
}
|
|
if name == assign.PoliciesBucket && err != nil {
|
|
// The seat mesh-wants' state, new with it (ADR 0565 §7): bound once the bus says it was made; until then no
|
|
// person's policy is read, so the controller's beside each composition holds, as before.
|
|
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
|
continue
|
|
}
|
|
if (name == assign.FindingsBucket || name == takeoverBucket) && err != nil {
|
|
// New with health leaving the controller (issue 1952 WM6): without them the conditions are said straight to the
|
|
// router and kept as no finding, as before.
|
|
logf("the bucket %s cannot be opened (%v): the assigner's conditions are said straight to the router until it starts again", name, err)
|
|
continue
|
|
}
|
|
if sourceOfGrants(name) && err != nil {
|
|
// The grants part's grants and the controller's lease choose where grants are read from (issue 774): without
|
|
// them the controller's grants are read, as before the lease is handed away.
|
|
logf("the bucket %s cannot be opened (%v): the controller's grants are read until the assigner starts again", name, err)
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err)
|
|
}
|
|
out[name] = kv
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// startsWithout says the assigner starts without the bucket when it cannot be opened (open says why for each): the seat's
|
|
// states and those new with a delivery of their own, bound once made, and the two that choose the grants' source.
|
|
func startsWithout(name string) bool {
|
|
switch name {
|
|
case delivery.SeatHoldersBucket, keptBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket, takeoverBucket,
|
|
delivery.AssignerPutsBucket:
|
|
return true
|
|
}
|
|
return slices.Contains(seatBuckets, name) || sourceOfGrants(name)
|
|
}
|
|
|
|
// sourceOfGrants says the bucket is the grants part's grants or the controller's lease: neither stops the assigner when it
|
|
// is not there.
|
|
func sourceOfGrants(name string) bool {
|
|
return name == delivery.PartGrantsBucket || name == buscompose.LeaseBucket
|
|
}
|
|
|
|
// run is the program: it returns when ctx ends or the bus cannot be used.
|
|
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
|
|
logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags)
|
|
account, err := process.ReadBrokerAccount(getenv)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
conn, js, err := account.Dial()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer conn.Close()
|
|
kvs, err := open(ctx, js, logger.Printf)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot
|
|
// be said now: process.Conn says it again at every reconnect.
|
|
said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer said.Close()
|
|
if err := said.Keeps(keptStates...); err != nil {
|
|
logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
|
|
}
|
|
thisNode = account.Node
|
|
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
|
|
findings := findingsOf(ctx, kvs, logger.Printf)
|
|
buildLog := func(handle func(stdio.LogEntry) error) error {
|
|
return said.ReadLog(delivery.BuilderModule, delivery.BuilderLog).Watch("", stdio.LogFromAll, handle)
|
|
}
|
|
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, buildLog, logger)
|
|
}
|
|
|
|
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, and the
|
|
// condition keeper's take-over key, watched, so a condition is said straight to the router only until the take-over. A
|
|
// bucket it cannot open is said: without its findings the conditions are said straight, as before; without the take-over
|
|
// key they are said straight too, the controller being followed.
|
|
func findingsOf(ctx context.Context, kvs buckets, logf func(string, ...any)) assign.Findings {
|
|
var f assign.Findings
|
|
if kv := kvs[assign.FindingsBucket]; kv != nil {
|
|
f.Put = func(key string, value []byte) error { _, err := kv.Put(ctx, key, value); return err }
|
|
f.Delete = func(key string) error { return kv.Delete(ctx, key) }
|
|
f.Keys = func() ([]string, error) {
|
|
keys, err := kv.Keys(ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return nil, nil
|
|
}
|
|
return keys, err
|
|
}
|
|
}
|
|
var keeper atomic.Bool
|
|
f.ToKeeper = keeper.Load
|
|
takeover := takeoverBucket
|
|
kv := kvs[takeover]
|
|
if kv == nil {
|
|
return f
|
|
}
|
|
w, err := kv.Watch(ctx, health.TakeoverKey)
|
|
if err != nil {
|
|
process.NotWatched(logf, takeover, err, "the controller is followed: the conditions are said straight to the router")
|
|
return f
|
|
}
|
|
go func() {
|
|
defer w.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
logf("the watch of %s ended: the last source it named is followed until the assigner starts again", takeover)
|
|
return
|
|
}
|
|
if e == nil {
|
|
continue
|
|
}
|
|
var raw []byte
|
|
if e.Operation() == jetstream.KeyValuePut {
|
|
raw = e.Value()
|
|
}
|
|
t, err := health.DecodeTakeover(raw, raw != nil)
|
|
if err != nil {
|
|
logf("the take-over key cannot be read (%v): the controller is followed", err)
|
|
}
|
|
if was := keeper.Swap(t.Follow == health.FollowKeeper); was != keeper.Load() {
|
|
logf("the assigner's conditions are said by %s from now (the take-over key follows %s)", t.Seat(), t.Follow)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return f
|
|
}
|
|
|
|
// thisNode is the node this process runs on, as its credential names it: said with each of the seat's events.
|
|
var thisNode string
|
|
|
|
// emitting is the process connection as the puts' caller: its calls, and its events with no options.
|
|
type emitting struct{ *process.Conn }
|
|
|
|
func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(eventType, body) }
|
|
|
|
// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and
|
|
// watched only for keys, so a manifest that arrives wakes a composition that waits for it.
|
|
var watched = []string{
|
|
buscompose.ResolvedBucket,
|
|
provider.AssignmentsBucket,
|
|
delivery.PartGrantsBucket,
|
|
buscompose.LeaseBucket,
|
|
delivery.ControllerComposedBucket,
|
|
delivery.AnswersBucket,
|
|
delivery.BuilderBuildsBucket,
|
|
delivery.CandidatesBucket,
|
|
delivery.ModulesBucket,
|
|
delivery.PlacementBucket,
|
|
delivery.PlacementProposalsBucket,
|
|
delivery.PlacementCandidatesBucket,
|
|
delivery.PlacementPulsesBucket,
|
|
delivery.PresenceBucket,
|
|
delivery.HaveBucket,
|
|
delivery.AssignerAsksBucket,
|
|
delivery.AssignerCompositionsBucket,
|
|
delivery.AssignerDifferencesBucket,
|
|
delivery.AssignerWantsBucket,
|
|
assign.PoliciesBucket,
|
|
provisions.ProvisionsBucket,
|
|
keptBucket,
|
|
delivery.SeatHoldersBucket,
|
|
}
|
|
|
|
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
|
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
|
|
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, buildLog logWatch, logger *log.Logger) error {
|
|
w := newWorld()
|
|
dirty := make(chan struct{}, 1)
|
|
wake := func() {
|
|
select {
|
|
case dirty <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
// The builder's log (ADR 0571 clause 1): every entry it keeps before the first pass, then each new one as it is appended;
|
|
// a module's newer done trunk build makes a pass, as a composition waiting on its commit is judged then. Where it cannot be
|
|
// read, a composition names no running reader of the manifests until the assigner starts again, and it is said.
|
|
if buildLog != nil {
|
|
var first atomic.Bool
|
|
if err := buildLog(func(le stdio.LogEntry) error {
|
|
if w.putBuilt(le.Value) && first.Load() {
|
|
wake()
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
logger.Printf("%s is not read (%v): no composition is judged against a running reader of the manifests until the assigner starts again",
|
|
delivery.BuildLogRead, err)
|
|
}
|
|
first.Store(true)
|
|
}
|
|
errs := make(chan error, len(watched)+2)
|
|
settled := make(chan struct{}, len(watched)+2)
|
|
watch := func(name string, opts ...jetstream.WatchOpt) error {
|
|
kw, err := kvs[name].WatchAll(ctx, opts...)
|
|
if err != nil {
|
|
return fmt.Errorf("the bucket %s cannot be watched: %w", name, err)
|
|
}
|
|
// The puts' watch never stops the wants (ADR 0546): when it ends, the puts are no longer heard, said in the journal.
|
|
optional := name == delivery.AssignerPutsBucket
|
|
go func() {
|
|
defer kw.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
errs <- nil
|
|
return
|
|
case e, ok := <-kw.Updates():
|
|
if !ok {
|
|
if optional {
|
|
logger.Printf("the watch of %s ended: no put is taken further until the assigner starts again", name)
|
|
return
|
|
}
|
|
errs <- fmt.Errorf("the watch of %s ended", name)
|
|
return
|
|
}
|
|
if e == nil {
|
|
if !optional { // the puts never gate the first pass, so never the wants (ADR 0546)
|
|
settled <- struct{}{}
|
|
}
|
|
continue
|
|
}
|
|
var value []byte
|
|
if e.Operation() == jetstream.KeyValuePut {
|
|
value = e.Value()
|
|
}
|
|
before := w.handedAway()
|
|
w.putEntry(name, e.Key(), value, e.Revision(), e.Created())
|
|
if name == buscompose.LeaseBucket && w.handedAway() == before {
|
|
// The controller renews its lease every few seconds: only the lease's being handed away makes a pass.
|
|
continue
|
|
}
|
|
wake()
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
// The puts' watch is not waited for before the first pass: a watch of them that ends before it handed what stands only
|
|
// stops the puts, never the wants.
|
|
if kvs[delivery.AssignerPutsBucket] != nil {
|
|
if err := watch(delivery.AssignerPutsBucket); err != nil {
|
|
logger.Printf("%v: no put is taken further until it is opened again", err)
|
|
delete(kvs, delivery.AssignerPutsBucket)
|
|
}
|
|
}
|
|
// The puts' bucket is made by the controller from the manifest's states at its start (ADR 0201), and the assigner's
|
|
// credential reads it once the bus's user list carries the grant: neither waits for the assigner. So while it is not open,
|
|
// a pass started by a change opens it again, at most once a minute, and from then on the puts are watched and taken
|
|
// further, without a restart.
|
|
var triedPuts time.Time
|
|
reopenPuts := func(p *passer) {
|
|
if kvs[delivery.AssignerPutsBucket] != nil || time.Since(triedPuts) < PutsReopenEvery {
|
|
return
|
|
}
|
|
triedPuts = time.Now()
|
|
kv, err := js.KeyValue(ctx, delivery.AssignerPutsBucket)
|
|
if err != nil {
|
|
return
|
|
}
|
|
kvs[delivery.AssignerPutsBucket] = kv
|
|
if err := watch(delivery.AssignerPutsBucket); err != nil {
|
|
logger.Printf("%v: no put is taken further until it is opened again", err)
|
|
delete(kvs, delivery.AssignerPutsBucket)
|
|
return
|
|
}
|
|
if bus != nil {
|
|
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
|
|
}
|
|
logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
|
|
}
|
|
started := 0
|
|
for _, name := range watched {
|
|
if kvs[name] == nil && (startsWithout(name) || name == provider.AssignmentsBucket) {
|
|
continue // not on the bus at this start (open said so)
|
|
}
|
|
started++
|
|
var opts []jetstream.WatchOpt
|
|
if name == delivery.HaveBucket {
|
|
opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
|
|
}
|
|
if err := watch(name, opts...); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
|
|
return err
|
|
}
|
|
// The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is
|
|
// bound once the bus says it was created, and watched from then on, with no restart (issue 1790's kind). Until then the
|
|
// controller's assignments alone name where modules run (ADR 0549 §5).
|
|
type appeared struct {
|
|
name string
|
|
kv jetstream.KeyValue
|
|
}
|
|
placementAppeared := make(chan appeared, len(seatBuckets)+2)
|
|
for _, name := range seatBuckets {
|
|
if kvs[name] == nil && js != nil {
|
|
process.AwaitBucket(ctx, js, nc, name, "the seat mesh-placement waits for it",
|
|
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{name, kv} })
|
|
}
|
|
}
|
|
// And the two that choose where grants are read from, with the same delivery as the grants part that writes one.
|
|
for _, name := range []string{delivery.PartGrantsBucket, buscompose.LeaseBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, delivery.SeatHoldersBucket} {
|
|
if kvs[name] == nil && js != nil {
|
|
name := name
|
|
process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read",
|
|
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{name, kv} })
|
|
}
|
|
}
|
|
takePlacement := func(a appeared) error {
|
|
if kvs[a.name] != nil {
|
|
return nil // bound already, by the other of its two events
|
|
}
|
|
kvs[a.name] = a.kv
|
|
logger.Printf("the bucket %s is open and watched", a.name)
|
|
return watch(a.name)
|
|
}
|
|
for range started + 1 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case err := <-errs:
|
|
return err
|
|
case <-settled:
|
|
}
|
|
}
|
|
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}}
|
|
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
|
|
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
|
|
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
|
|
}
|
|
if bus != nil {
|
|
p.seat = newSeatActs(bus, wake, logger.Printf, w.handedAway)
|
|
}
|
|
if js != nil {
|
|
p.publish = streamPublisher(ctx, js, thisNode)
|
|
}
|
|
wake()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case err := <-errs:
|
|
return err
|
|
case <-settled:
|
|
case kv := <-placementAppeared:
|
|
if err := takePlacement(kv); err != nil {
|
|
return err
|
|
}
|
|
case <-dirty:
|
|
reopenPuts(p)
|
|
// The grants part makes the seat's buckets in the pass that seals the list granting them, so their creation can
|
|
// be said before this credential may hear it: every pass, itself started by a watched change, binds each one the
|
|
// bus now holds, and says the seat waits while one is missing.
|
|
var unbound []string
|
|
for _, name := range seatBuckets {
|
|
if kvs[name] != nil || js == nil {
|
|
continue
|
|
}
|
|
if kv, err := js.KeyValue(ctx, name); err == nil {
|
|
if err := takePlacement(appeared{name, kv}); err != nil {
|
|
return err
|
|
}
|
|
continue
|
|
}
|
|
unbound = append(unbound, name)
|
|
}
|
|
p.sayWaits(unbound)
|
|
p.pass(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
// PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open.
|
|
var PutsReopenEvery = time.Minute
|
|
|
|
// sayVia publishes a condition on the operator-channel seat's one subject for conditions, under the assigner's own name
|
|
// (ADR 0437 clause 9).
|
|
func sayVia(ctx context.Context, js jetstream.JetStream) func(asks.Condition) error {
|
|
return func(c asks.Condition) error {
|
|
subject, err := asks.ConditionSubject(delivery.AssignerModule)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
data, err := asks.EncodeCondition(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = js.Publish(ctx, subject, data)
|
|
return err
|
|
}
|
|
}
|