From e0b4f73c9eb07b03f2590580d356b0e7e2bdc915 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:59:18 +0100 Subject: [PATCH] RN17 (fill): the runner says its stores' fill and a full store refuses a start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The seat gains the events filling, full and room; at each step's end and each start the runner reads the fill of pipelines, step-outputs, kept and step-log against their caps, says filling past 80 %, full at the cap and room below 70 %, and a start while a store is full is refused, said in the journal and heard again by the bus. Builds: design 47 G §366, §181 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../go/seats/meshpipelinerunner/contract.go | 23 +++- .../seats/mesh-pipeline-runner/seat.json | 18 +++ .../ts/seats/meshpipelinerunner/index.d.ts | 18 +++ .../ts/seats/meshpipelinerunner/index.js | 9 ++ .../cmd/mesh-delivery-pipeline/aged.go | 29 +++++ .../cmd/mesh-delivery-pipeline/main.go | 7 +- .../internal/pipeline/bounds.go | 1 + .../internal/pipeline/fill.go | 113 ++++++++++++++++++ .../internal/pipeline/fill_test.go | 52 ++++++++ .../internal/pipeline/start.go | 5 + .../internal/pipeline/steps.go | 3 + 11 files changed, 276 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/fill.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 6c2954b3..8cadf686 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -38,6 +38,9 @@ const ( EventStepEnded = "step-ended" EventWaitHeld = "wait-held" EventWaitEnded = "wait-ended" + EventFilling = "filling" + EventFull = "full" + EventRoom = "room" ) // Verbs is every verb the seat serves, in its definition's order: a new copy each call. @@ -47,7 +50,7 @@ func Verbs() []string { // Events is every event published on the seat's subject: a new copy each call. func Events() []string { - return []string{EventPipelineStarted, EventPipelineEnded, EventGroupEnded, EventRunStarted, EventStepStarted, EventStepEnded, EventWaitHeld, EventWaitEnded} + return []string{EventPipelineStarted, EventPipelineEnded, EventGroupEnded, EventRunStarted, EventStepStarted, EventStepEnded, EventWaitHeld, EventWaitEnded, EventFilling, EventFull, EventRoom} } // Family is where the seat's events are kept: the shared events stream, filtered to the seat's subjects. @@ -144,6 +147,15 @@ func SubjectWaitEnded(module, branch, commitOrGroup string) string { // SubjectWaitEndedAll hears every wait-ended. const SubjectWaitEndedAll = "mesh.seat.mesh-pipeline-runner.event.wait-ended.*.*.*" +// SubjectFilling is where filling is published. +const SubjectFilling = "mesh.seat.mesh-pipeline-runner.event.filling" + +// SubjectFull is where full is published. +const SubjectFull = "mesh.seat.mesh-pipeline-runner.event.full" + +// SubjectRoom is where room is published. +const SubjectRoom = "mesh.seat.mesh-pipeline-runner.event.room" + const ( StatePipelines = "pipelines" StateKept = "kept" @@ -978,6 +990,15 @@ type WaitHeldEvent = Wait // WaitEndedEvent is the body of the event wait-ended. type WaitEndedEvent = Wait +// FillingEvent is the body of the event filling. +type FillingEvent = Fill + +// FullEvent is the body of the event full. +type FullEvent = Fill + +// RoomEvent is the body of the event room. +type RoomEvent = Fill + // PipelinesByBranchRevisionModuleValue is one value of the state pipelines under the key ... type PipelinesByBranchRevisionModuleValue = Pipeline diff --git a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json index fcb5b4e8..4d82a907 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -591,6 +591,24 @@ "max-age-days": 90, "max-mib": 32 } + }, + { + "name": "filling", + "body": { + "$ref": "#/defs/Fill" + } + }, + { + "name": "full", + "body": { + "$ref": "#/defs/Fill" + } + }, + { + "name": "room", + "body": { + "$ref": "#/defs/Fill" + } } ], "family": { diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts index f1834d9d..579c3a87 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -28,6 +28,9 @@ export declare const events: { readonly stepEnded: "step-ended"; readonly waitHeld: "wait-held"; readonly waitEnded: "wait-ended"; + readonly filling: "filling"; + readonly full: "full"; + readonly room: "room"; }; export declare const family: { readonly name: "mesh-pipeline-runner.events"; readonly maxAgeDays: 7 }; @@ -80,6 +83,12 @@ export declare function waitEndedSubject(module: string, branch: string, commitO export declare const waitEndedSubjectAll: "mesh.seat.mesh-pipeline-runner.event.wait-ended.*.*.*"; +export declare const fillingSubject: "mesh.seat.mesh-pipeline-runner.event.filling"; + +export declare const fullSubject: "mesh.seat.mesh-pipeline-runner.event.full"; + +export declare const roomSubject: "mesh.seat.mesh-pipeline-runner.event.room"; + export declare const states: { readonly pipelines: "pipelines"; readonly kept: "kept"; @@ -597,6 +606,15 @@ export type WaitHeldEvent = Wait; /** is the body of the event wait-ended. */ export type WaitEndedEvent = Wait; +/** is the body of the event filling. */ +export type FillingEvent = Fill; + +/** is the body of the event full. */ +export type FullEvent = Fill; + +/** is the body of the event room. */ +export type RoomEvent = Fill; + /** is one value of the state pipelines under the key ... */ export type PipelinesByBranchRevisionModuleValue = Pipeline; diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js index a36f83ae..e797aedb 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js @@ -28,6 +28,9 @@ export const events = { stepEnded: "step-ended", waitHeld: "wait-held", waitEnded: "wait-ended", + filling: "filling", + full: "full", + room: "room", }; export const family = { name: "mesh-pipeline-runner.events", maxAgeDays: 7 }; @@ -96,6 +99,12 @@ export function waitEndedSubject(module, branch, commitOrGroup) { export const waitEndedSubjectAll = "mesh.seat.mesh-pipeline-runner.event.wait-ended.*.*.*"; +export const fillingSubject = "mesh.seat.mesh-pipeline-runner.event.filling"; + +export const fullSubject = "mesh.seat.mesh-pipeline-runner.event.full"; + +export const roomSubject = "mesh.seat.mesh-pipeline-runner.event.room"; + export const states = { pipelines: "pipelines", kept: "kept", diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go index 7d69152b..e0dca28d 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go @@ -146,3 +146,32 @@ func newAger(ctx context.Context, js jetstream.JetStream, cred process.Account, type pipelineDeleter struct{ pipelineStore } func (d pipelineDeleter) Delete(key string) error { return d.kv.Delete(d.ctx, key) } + +// storeFills reads the fill of the runner's stores against their caps: pipelines and kept (their key-value streams), step-outputs +// (its object store's) and step-log (its log's), as the bus's own streams say it (design 47 G §366, §486). +func storeFills(ctx context.Context, js jetstream.JetStream, cred process.Account, now func() time.Time) func() ([]seat.Fill, error) { + streams := []struct{ store, stream string }{ + {seat.StatePipelines, "KV_" + ownBucket(cred.Module, seat.StatePipelines)}, + {seat.StateKept, "KV_" + ownBucket(cred.Module, seat.StateKept)}, + {seat.StateStepOutputs, "OBJ_" + catalogue.ObjectsBucket(cred.Module, seat.StateStepOutputs)}, + {seat.StateStepLog, buscompose.LogStreamName(cred.Module, seat.StateStepLog)}, + } + return func() ([]seat.Fill, error) { + var out []seat.Fill + var first error + for _, s := range streams { + st, err := js.Stream(ctx, s.stream) + if err == nil { + var info *jetstream.StreamInfo + if info, err = st.Info(ctx); err == nil { + at := now().UTC() + out = append(out, seat.Fill{Store: s.store, Bytes: int64(info.State.Bytes), Cap: info.Config.MaxBytes, At: &at}) + } + } + if err != nil && first == nil { + first = fmt.Errorf("the fill of %s is not read: %w", s.store, err) + } + } + return out, first + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 44edf54a..24216b05 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -13,6 +13,7 @@ import ( "os/signal" "sync/atomic" "syscall" + "time" "github.com/nats-io/nats.go" @@ -106,13 +107,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearer := newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf) hearer.Asker = busAsker{conn: conn, cred: cred} hearer.Ager = newAger(ctx, js, cred, pipelines, kept, aged, logger.Printf) + // The stores' fill is said before a store is full, and a full one refuses a start (§366). + fills := &pipeline.Fills{Read: storeFills(ctx, js, cred, time.Now), Logf: logger.Printf, + Say: func(event string, f seat.Fill) error { return say("mesh.seat."+seat.Seat+".event."+event, f) }} + hearer.Fills = fills hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask) hearer.Failures = &pipeline.Failures{Findings: rv.Findings, Observer: cred.Module, Author: commitAuthor(calls.Call), Logf: logger.Printf} // Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked // for its step. starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say, Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, - Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} + Arm: armBound(ctx, bounds), Fills: fills, Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go index 1830cff9..f23d4781 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go @@ -86,6 +86,7 @@ func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error { h.logEnd(key, *s, now) h.Failures.StepEnded(h.Store, p, *s) h.Ager.Age() + h.Fills.Check() if err := h.say(seat.EventStepEnded, p, s); err != nil { return fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err) } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/fill.go b/modules/mesh-delivery-pipeline/internal/pipeline/fill.go new file mode 100644 index 00000000..bee8d812 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/fill.go @@ -0,0 +1,113 @@ +package pipeline + +import ( + "errors" + "fmt" + "sync" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// The fill at which a store says `filling` (80 %, past it) and `room` (70 %, below it) on the seat (design 47 G §366, §181). +const ( + FillingPercent = 80 + RoomPercent = 70 +) + +// ErrFull is a start refused because a store of the runner is full (§366): said, and heard again by the bus. +var ErrFull = errors.New("a store of the runner is full") + +// Fills says each of the runner's stores' fill before it is full (design 47 G §366): `filling` past 80 % of the cap, `full` at +// the cap, and `room` below 70 % once it said either; a store full refuses a new pipeline, loudly, and nothing is dropped to make room. +type Fills struct { + // Read is each store's fill now: pipelines, step-outputs, kept and step-log. + Read func() ([]seat.Fill, error) + // Say says one of the seat's events, filling, full or room, with the store's fill. + Say func(event string, f seat.Fill) error + Logf func(string, ...any) + mu sync.Mutex + said map[string]string // the store's last event said, "" before the first check + full map[string]bool +} + +// Check reads each store's fill and says what changed since the last check; it is read at each step's end and at each start. +func (f *Fills) Check() { + if f == nil || f.Read == nil { + return + } + fills, err := f.Read() + if err != nil { + f.logf("the stores' fill is not read: %v", err) + } + f.mu.Lock() + defer f.mu.Unlock() + if f.said == nil { + f.said, f.full = map[string]string{}, map[string]bool{} + } + for _, s := range fills { + if s.Cap <= 0 { + continue + } + event := "" + switch last := f.said[s.Store]; { + case s.Bytes >= s.Cap: + event = seat.EventFull + case s.Bytes*100 > s.Cap*FillingPercent: + if last != seat.EventFilling && last != seat.EventFull { + event = seat.EventFilling + } + case s.Bytes*100 < s.Cap*RoomPercent: + // A store below 70 % says room once it said otherwise, and at the first check after a start: a keeper holding a + // finding of the store from before is told it cleared. + if last != seat.EventRoom { + event = seat.EventRoom + } + } + f.full[s.Store] = s.Bytes >= s.Cap + if event == "" || event == f.said[s.Store] { + continue + } + if err := f.Say(event, s); err != nil { + f.logf("%s of %s is not said: %v", event, s.Store, err) + continue + } + f.said[s.Store] = event + } +} + +// Full is the stores full now, as the last check read them. +func (f *Fills) Full() []string { + if f == nil { + return nil + } + f.mu.Lock() + defer f.mu.Unlock() + var out []string + for s, full := range f.full { + if full { + out = append(out, s) + } + } + return out +} + +func (f *Fills) logf(format string, args ...any) { + if f != nil && f.Logf != nil { + f.Logf(format, args...) + } +} + +// refuseStart is the error of a start while a store is full, naming each store; it is said in the journal, and the start is heard +// again by the bus, so the runner takes it up on room (§366). +func (f *Fills) refuseStart(key string) error { + if f == nil { + return nil + } + f.Check() + full := f.Full() + if len(full) == 0 { + return nil + } + f.logf("the start of %s is refused: %v full", key, full) + return fmt.Errorf("%w: %v; %s starts on room", ErrFull, full, key) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go new file mode 100644 index 00000000..84486f21 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go @@ -0,0 +1,52 @@ +package pipeline + +import ( + "errors" + "testing" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// A store past 80 % of its cap says `filling` once, at its cap `full` and refuses a start, said; below 70 % it says `room` and +// the start goes (design 47 G §366). +func TestAStoreFillingIsSaidAndAFullStoreRefusesAStart(t *testing.T) { + bytes := int64(10) + var said []string + var journal []string + f := &Fills{ + Read: func() ([]seat.Fill, error) { return []seat.Fill{{Store: "pipelines", Bytes: bytes, Cap: 100}}, nil }, + Say: func(event string, fl seat.Fill) error { said = append(said, event+":"+fl.Store); return nil }, + Logf: func(format string, a ...any) { journal = append(journal, format) }} + steps := []struct { + bytes int64 + say string + }{{10, "room:pipelines"}, {75, ""}, {81, "filling:pipelines"}, {85, ""}, {100, "full:pipelines"}, {75, ""}, {69, "room:pipelines"}} + for _, st := range steps { + bytes, said = st.bytes, nil + f.Check() + got := "" + if len(said) == 1 { + got = said[0] + } + if got != st.say || len(said) > 1 { + t.Fatalf("at %d of 100 bytes the runner said %v, want %q", st.bytes, said, st.say) + } + } + bytes = 100 + s := &Starter{Pipelines: failingCreate{}, Fills: f, Say: func(string, any) error { return nil }} + ok, err := s.Start(Begin{Branch: "main", Source: 1, Module: "app"}) + if ok || !errors.Is(err, ErrFull) { + t.Fatalf("a start with pipelines full: %v, %v; want it refused with ErrFull", ok, err) + } + if len(journal) == 0 { + t.Error("the refused start is not said in the journal") + } + bytes = 10 + if ok, err := s.Start(Begin{Branch: "main", Source: 1, Module: "app"}); !ok || err != nil { + t.Fatalf("a start on room: %v, %v", ok, err) + } +} + +type failingCreate struct{} + +func (failingCreate) Create(string, []byte) error { return nil } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 9e5737aa..ad70f5b9 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -70,6 +70,8 @@ type Starter struct { Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm + // Fills refuses a start while a store of the runner is full (§366); nil refuses none. + Fills *Fills } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -83,6 +85,9 @@ func (s *Starter) Start(b Begin) (bool, error) { now = s.Now().UTC() } key := Key(b.Branch, b.Source, b.Module) + if err := s.Fills.refuseStart(key); err != nil { + return false, err + } p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp, Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now, Steps: []seat.StepResult{}} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go index 821ef52f..45c50d41 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go @@ -111,6 +111,8 @@ type Hearer struct { Failures *Failures // Ager ages the runner's own kept data at each step's end (§330); nil ages none. Ager *Ager + // Fills says the stores' fill at each step's end (§366); nil says none. + Fills *Fills } func (h *Hearer) now() time.Time { @@ -197,6 +199,7 @@ func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pi h.logEnd(ev.Pipeline, *s, now, keptKeys...) h.Failures.StepEnded(h.Store, p, *s) h.Ager.Age() + h.Fills.Check() if err := h.say(seat.EventStepEnded, p, s); err != nil { return p, fmt.Errorf("pipeline %s's %s is written but not said: %w", ev.Pipeline, seat.EventStepEnded, err) }