jschoubben
80a0183567
The store's nightly collector deleted the bare blobs of node-tools', mesh-host's and the forge bridge's done keys, which the controller had never taken in, and the builder kept them done: their next delivery would refuse. The builder now asks the controller's held before it leaves a done trunk key done (once per key per process, at its start and on each module value), claims a key whose blob is gone building as a new revision and says so in its log and the key's step, and hears artifacts-gone, said when a take-in found a gone blob. No timer: a start, a value or an event.
62 lines
2.2 KiB
Go
62 lines
2.2 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
|
|
"mesh-delivery-builder/internal/build"
|
|
)
|
|
|
|
// eventsStream is the stream every module's and role's events are on, where the controller makes the builder's durable
|
|
// consumer of artifacts-gone (mesh-controller internal/broker EventsStream).
|
|
const eventsStream = "EVENTS"
|
|
|
|
// 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.
|
|
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})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("asking %s: %w", delivery.HeldVerb, err)
|
|
}
|
|
a, err := delivery.DecodeHeldAnswer(raw, references)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(a.Gone) == 0 && len(a.Unknown) > 0 {
|
|
var why []string
|
|
for r, w := range a.Unknown {
|
|
why = append(why, r+": "+w)
|
|
}
|
|
sort.Strings(why)
|
|
return nil, fmt.Errorf("the artifact store could not be asked about %s", strings.Join(why, "; "))
|
|
}
|
|
return a.Gone, nil
|
|
}
|
|
}
|
|
|
|
// heardGone takes one artifacts-gone event in: acknowledged, then handed to the builder, which asks the store and builds
|
|
// the key again. A body that is not one is said and acknowledged, so it is not handed over again.
|
|
func heardGone(ctx context.Context, b *build.Builder, m jetstream.Msg, logf func(string, ...any)) {
|
|
e, err := delivery.DecodeArtifactsGone(m.Data())
|
|
if ackErr := m.Ack(); ackErr != nil {
|
|
logf("acknowledging %s: %v", m.Subject(), ackErr)
|
|
}
|
|
if err != nil {
|
|
logf("%s holds what is not %s, left: %v", m.Subject(), delivery.ArtifactsGoneEvent, err)
|
|
return
|
|
}
|
|
b.ArtifactsGone(ctx, e)
|
|
}
|