RN17 (fill): the runner says its stores' fill and a full store refuses a start

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

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 <branch>.<revision>.<module>.
 type PipelinesByBranchRevisionModuleValue = Pipeline
 

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": {

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 <branch>.<revision>.<module>. */
 export type PipelinesByBranchRevisionModuleValue = Pipeline;
 

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",

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
+	}
+}

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 {

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)
 		}

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)
+}

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 }

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{}}

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)
 		}