Builder: write each build's start and end to its own log on the bus, and ask the seat artifact-store's held and hear its artifacts-gone, the controller's only until the lease write, so build history and the store's sweep leave the controller (issue 1955, ADR 0571 §1, §2)
Pins mesh-sdk f9a496d.
This commit is contained in:
@@ -120,6 +120,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Manifests: manifestReader{manifests},
|
||||
Logf: logger.Printf,
|
||||
Stored: storedVia(calls.Call),
|
||||
Node: cred.Node,
|
||||
}
|
||||
// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
|
||||
// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
|
||||
@@ -129,6 +130,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf}
|
||||
trunks.start()
|
||||
b.Trunk = trunks.head
|
||||
// The builder's log (ADR 0571 §1): each build's start and end, on its own stream, read by the view, the indexer, the
|
||||
// assigner and mesh-catalog.
|
||||
b.Log = func(e delivery.BuildLogEntry) {
|
||||
if err := emitLog(context.WithoutCancel(ctx), js, cred, e); err != nil {
|
||||
logger.Printf("%v", err)
|
||||
}
|
||||
}
|
||||
b.Said = func(e build.AskedAgain) {
|
||||
if err := emitAskedAgain(context.WithoutCancel(ctx), js, cred, e); err != nil {
|
||||
logger.Printf("%v", err)
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
@@ -138,3 +139,25 @@ func announcedAgain(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e
|
||||
logf("%s.%s: the answer could not be written: %v", build.ReAsksState, e.Key(), err)
|
||||
}
|
||||
}
|
||||
|
||||
// emitLog says one entry of the builder's log on its stream `log` (ADR 0571 §1), under an id the stream keeps one of.
|
||||
func emitLog(ctx context.Context, js jetstream.JetStream, cred process.Account, e delivery.BuildLogEntry) error {
|
||||
data, err := delivery.EncodeBuildLogEntry(e)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, e.Event))
|
||||
id := e.Event + "." + e.Module + "." + e.FP + "." + e.At.Format(time.RFC3339Nano)
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", cred.Module)
|
||||
msg.Header.Set("x-node", cred.Node)
|
||||
msg.Header.Set("x-time", e.At.Format(time.RFC3339Nano))
|
||||
msg.Header.Set("content-type", "application/json")
|
||||
msg.Data = data
|
||||
pub, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil {
|
||||
return fmt.Errorf("logging %s of %s at %s: %w", e.Event, e.Module, e.FP, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -9,7 +9,9 @@ import (
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/handacts"
|
||||
|
||||
"mesh-delivery-builder/internal/build"
|
||||
)
|
||||
@@ -18,17 +20,27 @@ import (
|
||||
// consumer of artifacts-gone (mesh-controller internal/broker EventsStream).
|
||||
const eventsStream = "EVENTS"
|
||||
|
||||
// SeatHeld is the seat artifact-store's verb held (ADR 0571 §2).
|
||||
const SeatHeld = "seat:" + catalogue.ArtifactStoreSeat + ".held"
|
||||
|
||||
// toolCaller calls a tool or a seat verb as the builder's own account (mesh-sdk process.Conn.Call).
|
||||
type toolCaller func(key string, body any) (json.RawMessage, error)
|
||||
|
||||
// storedVia is build.Builder's Stored: it asks the controller's verb held (mesh-sdk delivery.HeldVerb; novox/hq issue
|
||||
// 1937) and answers the references the store no longer holds. One it could not ask about, with none gone, is an error:
|
||||
// the key is then left done, and the controller's take-in still refuses a gone blob.
|
||||
//
|
||||
// It asks the seat artifact-store's `held` (novox/hq ADR 0571 §2), and the controller's while that seat has no holder that
|
||||
// answers (a named exception of clause 10, until the lease write).
|
||||
func storedVia(call toolCaller) func(ctx context.Context, references []string) ([]string, error) {
|
||||
return func(_ context.Context, references []string) ([]string, error) {
|
||||
raw, err := call(delivery.HeldVerb, map[string]any{"references": references})
|
||||
body := map[string]any{"references": references}
|
||||
raw, err := call(SeatHeld, body)
|
||||
if handacts.NoResponder(err) {
|
||||
raw, err = call(delivery.HeldVerb, body)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("asking %s: %w", delivery.HeldVerb, err)
|
||||
return nil, fmt.Errorf("asking held: %w", err)
|
||||
}
|
||||
a, err := delivery.DecodeHeldAnswer(raw, references)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user