jschoubben
4628ba26b4
mesh/delivery merged: source.main moved to novox/mesh-delivery@db53ec975f01
mesh/delivery-group group fix/1945-ask-again-and-why merged: every member is merged
mesh/check pass 4628ba26b41a on 73be3df28cd4 | unit 167 run, 443 kept | integration 177 run, 509 kept | composition passed on 3 machines
The open helpers were copied into seven parts; code shared between modules lives in mesh-sdk, so they are mesh-sdk process's now and the parts are pinned at fix/1945-optional-buckets (e0d1de04), moved to its main once it merges, one delivery group with mesh-sdk.
79 lines
2.8 KiB
Go
79 lines
2.8 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** (mesh-sdk process.EssentialBucket, OptionalBucket) (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.
|
|
|
|
// 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)
|
|
}
|