CAT-NR1: every build narrates its asks of a seat in one place
asked, waiting on the seat every five minutes, stored or refused, as the ask's phase on its step key, so a publish or a copy that stands still is readable on the bus. PublishOne and PublishFiles narrate for every build module; the image copy and the late publish use the same narration. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -144,7 +144,10 @@ builder writes the older form beside it while a counted node-engine reads only t
|
||||
seat nothing holds, fails the step `not stored: <artifact>: <the holder's words>`. The packed bytes never stay on the node.
|
||||
A build the builder numbers later is published late: the module consumes the builder's `version.numbered` and, for a
|
||||
bundle or an archive with a digest, publishes that held digest under its version, naming its one part by digest and
|
||||
sending nothing (`latepublish.go`); a holder that no longer holds the digest fails the event, which is offered again.
|
||||
sending nothing (`latepublish.go`); a holder that no longer holds the digest fails the event, which is offered again. Every
|
||||
ask of a seat a build module makes (a publish, a late publish, an image's copy) is narrated by `buildstep/narrate.go`:
|
||||
asked, waiting on the seat every five minutes while it has not answered, stored or refused, each the ask's phase on its
|
||||
running step key (logged outside an ask).
|
||||
A contracts module (a package artifact naming `generated-from`) has its generated Go files regenerated with the seat
|
||||
generator this module is built with and compared (`buildstep/generated.go`): the unit level of `build_go_test` fails
|
||||
naming each file that differs, is missing or was not written by the generator (its `out-of-step`), and a build ask
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
package buildstep
|
||||
|
||||
// **A build narrates its asks of a seat** (CAT-NR1; design 51 *Publishing*): asked, waiting on the seat (said again every
|
||||
// NarrateEvery while it waits), stored, refused. Each line is the ask's phase (phase.go): logged with the ask's id and
|
||||
// written to its running step key, so the builder and the delivery view read from the bus where a publish or a copy
|
||||
// stands; outside an ask's run (a late publish, a tool called by hand) it is logged. One place for every build module:
|
||||
// PublishOne and PublishFiles narrate through it, and a step that waits on a seat's event (an image's copy) uses
|
||||
// NarrateSeat itself.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NarrateEvery is how often a build says again that it still waits on a seat.
|
||||
var NarrateEvery = 5 * time.Minute
|
||||
|
||||
// narrateTicks is the clock of the waiting lines; a test hands its own.
|
||||
var narrateTicks = func(every time.Duration) (<-chan time.Time, func()) {
|
||||
t := time.NewTicker(every)
|
||||
return t.C, t.Stop
|
||||
}
|
||||
|
||||
// SeatNarration is one ask of a build on a seat, said as it moves.
|
||||
type SeatNarration struct {
|
||||
ctx context.Context
|
||||
seat, what string
|
||||
asked time.Time
|
||||
tick <-chan time.Time
|
||||
stop func()
|
||||
}
|
||||
|
||||
// NarrateSeat says that the build asked seat for what (a verb and its object, such as "publish of nats/server") and
|
||||
// starts the waiting lines; Stored or Refused ends it.
|
||||
func NarrateSeat(ctx context.Context, seat, what string) *SeatNarration {
|
||||
n := &SeatNarration{ctx: ctx, seat: seat, what: what, asked: now()}
|
||||
n.tick, n.stop = narrateTicks(NarrateEvery)
|
||||
n.say("asked %s: %s", seat, what)
|
||||
return n
|
||||
}
|
||||
|
||||
// Waiting is the clock a waiting step selects on beside its outcome; on each tick it calls StillWaiting.
|
||||
func (n *SeatNarration) Waiting() <-chan time.Time { return n.tick }
|
||||
|
||||
// StillWaiting says that the build still waits on the seat, and since when.
|
||||
func (n *SeatNarration) StillWaiting() {
|
||||
n.say("waiting on %s: %s, %s since asked", n.seat, n.what, now().Sub(n.asked).Round(time.Second))
|
||||
}
|
||||
|
||||
// Stored says that the seat holds what the build asked it to keep (how: where, or what travelled).
|
||||
func (n *SeatNarration) Stored(how string) {
|
||||
n.stop()
|
||||
if how != "" {
|
||||
how = " (" + how + ")"
|
||||
}
|
||||
n.say("stored on %s: %s%s", n.seat, n.what, how)
|
||||
}
|
||||
|
||||
// Refused says that the seat did not keep it, and why.
|
||||
func (n *SeatNarration) Refused(err error) {
|
||||
n.stop()
|
||||
n.say("refused by %s: %s: %v", n.seat, n.what, err)
|
||||
}
|
||||
|
||||
func (n *SeatNarration) say(format string, a ...any) {
|
||||
line := fmt.Sprintf(format, a...)
|
||||
if r, ok := n.ctx.Value(phaseKey{}).(*phases); ok && r != nil {
|
||||
r.enter(line)
|
||||
return
|
||||
}
|
||||
log.Print(line)
|
||||
}
|
||||
|
||||
// OnSeat runs call as one narrated ask of the build on seat: asked, waiting every NarrateEvery while call runs, then
|
||||
// stored (how(answer)) or refused (its error).
|
||||
func OnSeat[T any](ctx context.Context, seat, what string, call func() (T, error), how func(T) string) (T, error) {
|
||||
n := NarrateSeat(ctx, seat, what)
|
||||
type answer struct {
|
||||
v T
|
||||
err error
|
||||
}
|
||||
done := make(chan answer, 1)
|
||||
go func() {
|
||||
v, err := call()
|
||||
done <- answer{v, err}
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
case a := <-done:
|
||||
if a.err != nil {
|
||||
n.Refused(a.err)
|
||||
} else {
|
||||
h := ""
|
||||
if how != nil {
|
||||
h = how(a.v)
|
||||
}
|
||||
n.Stored(h)
|
||||
}
|
||||
return a.v, a.err
|
||||
case <-n.Waiting():
|
||||
n.StillWaiting()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package buildstep
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// CAT-NR1: a build's publish on a seat says asked, waiting on the seat (on every tick of NarrateEvery while the holder
|
||||
// has not answered), and stored or refused, each as the ask's phase written to its running step key.
|
||||
func TestABuildNarratesItsPublishOnASeat(t *testing.T) {
|
||||
digest := "sha256:" + strings.Repeat("c", 64)
|
||||
for name, c := range map[string]struct {
|
||||
answer error
|
||||
ticks int
|
||||
want []string
|
||||
}{
|
||||
"stored": {want: []string{
|
||||
"asked go-package-registry: publish of mod 1.2.3 at " + digest,
|
||||
"stored on go-package-registry: publish of mod 1.2.3 at " + digest + " (0 of its 1 parts sent, the rest held)"}},
|
||||
"refused": {answer: errors.New("the holder is down"), want: []string{
|
||||
"asked go-package-registry: publish of mod 1.2.3 at " + digest,
|
||||
"refused by go-package-registry: publish of mod 1.2.3 at " + digest + ": go-package-registry's publish of mod: the holder is down"}},
|
||||
"waiting twice, then stored": {ticks: 2, want: []string{
|
||||
"asked go-package-registry: publish of mod 1.2.3 at " + digest,
|
||||
"waiting on go-package-registry: publish of mod 1.2.3 at " + digest + ", 5m0s since asked",
|
||||
"waiting on go-package-registry: publish of mod 1.2.3 at " + digest + ", 10m0s since asked",
|
||||
"stored on go-package-registry: publish of mod 1.2.3 at " + digest + " (0 of its 1 parts sent, the rest held)"}},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
clock := time.Date(2026, 11, 4, 16, 0, 0, 0, time.UTC)
|
||||
var mu sync.Mutex
|
||||
oldNow, oldTicks := now, narrateTicks
|
||||
now = func() time.Time { mu.Lock(); defer mu.Unlock(); return clock }
|
||||
tick := make(chan time.Time)
|
||||
stopped := make(chan struct{})
|
||||
narrateTicks = func(every time.Duration) (<-chan time.Time, func()) {
|
||||
if every != 5*time.Minute {
|
||||
t.Errorf("the waiting lines come every %s, want 5m", every)
|
||||
}
|
||||
return tick, func() { close(stopped) }
|
||||
}
|
||||
t.Cleanup(func() { now, narrateTicks = oldNow, oldTicks })
|
||||
|
||||
var said []string
|
||||
wrote := make(chan struct{}, 16)
|
||||
step := delivery.Step{ID: "a1", Tool: "build", Node: "ace", State: delivery.StepRunning, Started: clock}
|
||||
ph := &phases{key: "ace.a1", step: step, put: func(key string, value any) (uint64, error) {
|
||||
var s delivery.Step
|
||||
_ = json.Unmarshal(value.(json.RawMessage), &s)
|
||||
phasesMu.Lock()
|
||||
said = append(said, s.Phase)
|
||||
phasesMu.Unlock()
|
||||
wrote <- struct{}{}
|
||||
return 1, nil
|
||||
}}
|
||||
release := make(chan struct{})
|
||||
call := func(key string, body any) (json.RawMessage, error) {
|
||||
<-release
|
||||
if c.answer != nil {
|
||||
return nil, c.answer
|
||||
}
|
||||
return json.Marshal(reg.PublishAnswer{Held: true, Transfer: "t"})
|
||||
}
|
||||
ctx := context.WithValue(context.Background(), phaseKey{}, ph)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- PublishOne(ctx, call, GoSeat, "mod", "package", "1.2.3", digest, "t", []byte("x")) }()
|
||||
line := func() {
|
||||
select {
|
||||
case <-wrote:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the publish said no line")
|
||||
}
|
||||
}
|
||||
line() // asked, at the clock's start
|
||||
for i := 0; i < c.ticks; i++ {
|
||||
mu.Lock()
|
||||
clock = clock.Add(5 * time.Minute)
|
||||
mu.Unlock()
|
||||
tick <- clock
|
||||
line() // its waiting line
|
||||
}
|
||||
close(release)
|
||||
err := <-done
|
||||
if (err != nil) != (c.answer != nil) {
|
||||
t.Fatalf("PublishOne = %v", err)
|
||||
}
|
||||
<-stopped
|
||||
phasesMu.Lock()
|
||||
defer phasesMu.Unlock()
|
||||
// Each line is the step's phase, written to its key, in order.
|
||||
if fmt.Sprint(said) != fmt.Sprint(c.want) {
|
||||
t.Errorf("phases =\n%s\nwant\n%s", strings.Join(said, "\n"), strings.Join(c.want, "\n"))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
var phasesMu sync.Mutex
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
@@ -42,11 +43,29 @@ func TransferName(name, qualifier, digest string) string {
|
||||
// PublishOne publishes body, of digest, as name (at version, when it has one) of kind on seat, in one part.
|
||||
func PublishOne(ctx context.Context, call SeatCall, seat, name, kind, version, digest, transfer string, body []byte) error {
|
||||
in := reg.PublishInput{Name: name, Kind: kind, Version: version, Digest: digest, Transfer: transfer}
|
||||
_, err := publish.Publish(ctx, publish.Call(call), seat, in, []publish.Part{{Digest: digest, Size: int64(len(body)),
|
||||
Open: func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(body)), nil }}})
|
||||
_, err := OnSeat(ctx, seat, publishWhat(in), func() (publish.Sent, error) {
|
||||
return publish.Publish(ctx, publish.Call(call), seat, in, []publish.Part{{Digest: digest, Size: int64(len(body)),
|
||||
Open: func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(body)), nil }}})
|
||||
}, SentHow(1))
|
||||
return err
|
||||
}
|
||||
|
||||
// publishWhat names a publish in a narration: its name, its version when it has one, and its digest.
|
||||
func publishWhat(in reg.PublishInput) string {
|
||||
what := "publish of " + in.Name
|
||||
if in.Version != "" {
|
||||
what += " " + in.Version
|
||||
}
|
||||
return what + " at " + in.Digest
|
||||
}
|
||||
|
||||
// SentHow says what of a publish of parts travelled, for its stored line.
|
||||
func SentHow(parts int) func(publish.Sent) string {
|
||||
return func(s publish.Sent) string {
|
||||
return fmt.Sprintf("%d of its %d parts sent, the rest held", len(s.Parts), parts)
|
||||
}
|
||||
}
|
||||
|
||||
// File is one part of a publish kept on disk: an image's config, layer or manifest, by its digest and size.
|
||||
type File struct {
|
||||
Digest string
|
||||
@@ -61,5 +80,7 @@ func PublishFiles(ctx context.Context, call SeatCall, seat string, in reg.Publis
|
||||
for _, f := range files {
|
||||
ps = append(ps, publish.Part{Digest: f.Digest, Size: f.Size, Open: func() (io.ReadCloser, error) { return os.Open(f.Path) }})
|
||||
}
|
||||
return publish.Publish(ctx, publish.Call(call), seat, in, ps)
|
||||
return OnSeat(ctx, seat, publishWhat(in), func() (publish.Sent, error) {
|
||||
return publish.Publish(ctx, publish.Call(call), seat, in, ps)
|
||||
}, SentHow(len(files)))
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
@@ -48,8 +49,10 @@ func (m *Module) LatePublish(ctx context.Context, body []byte) (bool, error) {
|
||||
}
|
||||
in := registry.PublishInput{Name: n.Name, Kind: n.Kind, Version: n.Version, Tag: n.Tag, Digest: n.Digest,
|
||||
Fingerprint: n.Fingerprint, Commit: n.Commit, Transfer: transferName(n.Name, "v"+n.Version, n.Digest)}
|
||||
sent, err := publish.Publish(ctx, publish.Call(call), seat, in, []publish.Part{{Digest: n.Digest,
|
||||
Open: func() (io.ReadCloser, error) { return nil, errNotHere }}})
|
||||
sent, err := buildstep.OnSeat(ctx, seat, "late publish of "+n.Name+" "+n.Version+" at "+n.Digest, func() (publish.Sent, error) {
|
||||
return publish.Publish(ctx, publish.Call(call), seat, in, []publish.Part{{Digest: n.Digest,
|
||||
Open: func() (io.ReadCloser, error) { return nil, errNotHere }}})
|
||||
}, nil)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("the late publish of %s@%s (%s): %w", n.Name, n.Version, n.Digest, err)
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
@@ -106,8 +107,10 @@ func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
|
||||
return nil, fmt.Errorf("%s%s: keeping the copy's wait in %s: %v", delivery.NotStored, name, MirrorWaitsState, err)
|
||||
}
|
||||
defer m.Outcomes.drop(key, asked.Token)
|
||||
said := buildstep.NarrateSeat(ctx, buildstep.ImageSeat, "mirror of "+from)
|
||||
raw, err := m.call()("seat:"+buildstep.ImageSeat+"."+reg.VerbMirror, reg.MirrorInput{Name: module + "/" + name, Source: from, Digest: digest})
|
||||
if err != nil {
|
||||
said.Refused(err)
|
||||
return nil, fmt.Errorf("%s%s: %s's mirror: %v", delivery.NotStored, name, buildstep.ImageSeat, err)
|
||||
}
|
||||
var answer reg.MirrorAnswer
|
||||
@@ -121,13 +124,24 @@ func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
|
||||
for {
|
||||
// The state first: an outcome written there while this copy did not hear it is this copy's.
|
||||
if o, ok := m.Outcomes.kept(key); ok {
|
||||
return copyEnd(name, o, answer.Reference)
|
||||
res, err := copyEnd(name, o, answer.Reference)
|
||||
if err != nil {
|
||||
said.Refused(errors.New(o.Reason))
|
||||
} else {
|
||||
said.Stored(res.(CopyResult).Reference)
|
||||
}
|
||||
return res, err
|
||||
}
|
||||
select {
|
||||
case <-heard:
|
||||
case <-said.Waiting():
|
||||
said.StillWaiting()
|
||||
case <-bound:
|
||||
return nil, fmt.Errorf("%s%s: %s said no outcome of the copy of %s in %s", delivery.NotStored, name, buildstep.ImageSeat, from, wait)
|
||||
err := fmt.Errorf("%s said no outcome of the copy of %s in %s", buildstep.ImageSeat, from, wait)
|
||||
said.Refused(err)
|
||||
return nil, fmt.Errorf("%s%s: %w", delivery.NotStored, name, err)
|
||||
case <-ctx.Done():
|
||||
said.Refused(ctx.Err())
|
||||
return nil, fmt.Errorf("%s%s: stopped while waiting for the copy of %s: %w", delivery.NotStored, name, from, ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user