jschoubben
20027e53ff
Met twice on 2026-10-25: the builder of #48 exited on a refused read of the gate's source and crash-looped, and its new reasks bucket would have done the same. A grant reaches the bus with a delivery, so a refusal or a timeout at start is often only a grant not there yet; exiting for it stopped the whole part until a hand step on the bus. Every startup open in the parts is now essential or optional. Essential ones (a part's own state and the input it exists for) still stop the start, with an error naming the bucket and its grant. Optional ones are said and left: watches the bus refuses too. The builder builds trunk keys without the gate's source, which it opens again at the next candidate that asks a trunk head; toolchains and previews open again at their next event, as a missing bucket did. builder_ask_again says that asking a running step again builds it twice.
105 lines
4.0 KiB
Go
105 lines
4.0 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the builder of
|
|
// mesh-delivery #48 exited on a refused read of the gate's source, and its reasks bucket would have done the same). A grant
|
|
// reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted yet:
|
|
// exiting for it stops every build until a person's hand. A bucket the builder cannot build at all without (its builds,
|
|
// the indexer's modules and manifests) still stops the start, naming the grant to look at; every other is said and left,
|
|
// and opened again at the next event that needs it where there is one.
|
|
|
|
// essentialBucket opens a bucket the builder cannot work without; its error names the bucket and the grant.
|
|
func essentialBucket(ctx context.Context, js bucketOpener, name, does string) (jetstream.KeyValue, error) {
|
|
kv, err := js.KeyValue(ctx, name)
|
|
if err != nil {
|
|
return nil, essentialErr(name, does, err)
|
|
}
|
|
return kv, nil
|
|
}
|
|
|
|
// essentialErr is the error of a bucket the part cannot work without: it names the bucket and the grant.
|
|
func essentialErr(name, does string, err error) error {
|
|
return fmt.Errorf("the bucket %s cannot be read (%v): the builder cannot %s without it, so it stops; where the bus "+
|
|
"refused it or did not answer, this module's grant of %s (its manifest's reads or state, issued with its credential) "+
|
|
"has not reached the bus yet", name, err, does, name)
|
|
}
|
|
|
|
// optionalBucket opens a bucket the builder works without: any error is said with what is not heard, and nil answered.
|
|
func optionalBucket(ctx context.Context, js bucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue {
|
|
kv, err := js.KeyValue(ctx, name)
|
|
if err != nil {
|
|
logf("the bucket %s cannot be read (%v): %s", name, err, without)
|
|
return nil
|
|
}
|
|
return kv
|
|
}
|
|
|
|
// trunkSource is the gate's source history the candidates' previews are merged onto (sourceHeads), opened at start and,
|
|
// while it cannot be, again at the next candidate that asks a trunk head: a builder without it builds the trunk's keys,
|
|
// and a candidate's preview waits, saying why.
|
|
type trunkSource struct {
|
|
ctx context.Context
|
|
js bucketOpener
|
|
heads *sourceHeads
|
|
moved func(context.Context)
|
|
errs chan<- error
|
|
logf func(string, ...any)
|
|
|
|
mu sync.Mutex
|
|
opened bool
|
|
}
|
|
|
|
// ensure opens and settles the source's watch once, and follows it; an error leaves it closed for the next ask.
|
|
func (t *trunkSource) ensure() error {
|
|
t.mu.Lock()
|
|
defer t.mu.Unlock()
|
|
if t.opened {
|
|
return nil
|
|
}
|
|
kv, err := t.js.KeyValue(t.ctx, delivery.SourceBucket)
|
|
if err != nil {
|
|
return fmt.Errorf("the bucket %s cannot be read (%v)", delivery.SourceBucket, err)
|
|
}
|
|
w, err := kv.Watch(t.ctx, delivery.SourceMainKey, jetstream.IncludeHistory())
|
|
if err != nil {
|
|
return fmt.Errorf("the bucket %s cannot be watched (%v)", delivery.SourceBucket, err)
|
|
}
|
|
sw, err := settled(t.ctx, w, func(e jetstream.KeyValueEntry) { t.heads.put(e) })
|
|
if err != nil {
|
|
return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err)
|
|
}
|
|
t.opened = true
|
|
go func() {
|
|
t.errs <- follow(t.ctx, sw, func(e jetstream.KeyValueEntry) {
|
|
if t.heads.put(e) && t.moved != nil {
|
|
t.moved(t.ctx)
|
|
}
|
|
})
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
// start opens the source at start; one that cannot be is said, and the builder goes on.
|
|
func (t *trunkSource) start() {
|
|
if err := t.ensure(); err != nil {
|
|
t.logf("%v: the trunk's keys are built, and a candidate's preview waits until the source can be read, tried again at its next candidate", err)
|
|
}
|
|
}
|
|
|
|
// head is Builder.Trunk: the head of repository at source revision rev, opening the source first if it is not yet.
|
|
func (t *trunkSource) head(ctx context.Context, rev uint64, repository string) (string, error) {
|
|
if err := t.ensure(); err != nil {
|
|
return "", err
|
|
}
|
|
return t.heads.head(ctx, rev, repository)
|
|
}
|