novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

RN1: the pipeline runner's module, a foundation part with its kept data in old sections (issue 1966)

Browse Source 
The module mesh-delivery-pipeline claims the seat mesh-pipeline-runner, serving its twelve verbs,
asks no template, keeps pipelines, kept, aged, findings and its first views in old state sections,
its step log on an old stream (128 MiB) and its step outputs in an old objects section (1 GiB),
hears the events its pipelines start and end on, and needs the assigner's program delivered first.
Its program stops when its state pipelines cannot be read; its own module is its account's.

Builds: design 47 G §1, §21, §27, §59, §85, §95, §101, §130, §333
Not built, waiting on design answers: §76's health check (its tool's name), §85's caps of the old
state sections (no size field on an old state), step-log's 100 days (the validator's 30-day bound),
and §59's sent-objects (the objects template's cap the design names has no default).

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 11:32:31 +01:00
parent ab9b8d3471
commit cafff10266
5 changed files with 257 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go
+51
View File
@@ -0,0 +1,51 @@
// Command mesh-delivery-pipeline is the pipeline runner, holder of the seat mesh-pipeline-runner, a foundation part
// whose kept data are in its manifest's old sections (design 47 G §21, §27, §101). It names no module, branch or
// repository of its own (§7): its own module is the one its bus account names.
package main
import (
"context"
"fmt"
"io"
"log"
"os"
"os/signal"
"syscall"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-sdk/go/process"
)
func main() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
if err := run(ctx, os.Getenv, os.Stderr); err != nil {
fmt.Fprintln(os.Stderr, "pipeline runner:", err)
os.Exit(1)
}
}
// run is the program: it returns when ctx ends, or at once when the bus or the state pipelines cannot be used.
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
cred, err := process.ReadBrokerAccount(getenv)
if err != nil {
return err
}
logger := log.New(logOut, cred.Module+": ", log.LstdFlags)
conn, js, err := cred.Dial()
if err != nil {
return err
}
defer conn.Close()
// The state pipelines, `<module>_pipelines`, holds every pipeline by compare-and-set (§27): without it the
// runner keeps nothing, so it stops.
if _, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StatePipelines), "keep a pipeline"); err != nil {
return err
}
logger.Printf("holding %s", seat.Seat)
<-ctx.Done()
return nil
}
// ownBucket is a state of the runner's old state section on the bus, `<module>_<state>`.
func ownBucket(module, state string) string { return module + "_" + state }
modules/mesh-delivery-pipeline/go.mod
+17
View File
@@ -0,0 +1,17 @@
module mesh-delivery-pipeline
go 1.26.0
require (
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
github.com/nats-io/nats.go v1.54.0
)
require (
github.com/klauspost/compress v1.20.1 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/sys v0.48.0 // indirect
)
modules/mesh-delivery-pipeline/go.sum
+12
View File
@@ -0,0 +1,12 @@
github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
modules/mesh-delivery-pipeline/manifest_test.go
+77
View File
@@ -0,0 +1,77 @@
package pipeline_test
import (
"os"
"slices"
"testing"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// TestManifest holds the runner's manifest to its design lines, row by row.
func TestManifest(t *testing.T) {
raw, err := os.ReadFile("module.json")
if err != nil {
t.Fatal(err)
}
m, err := catalogue.Decode(raw)
if err != nil {
t.Fatalf("the manifest does not read: %v", err)
}
var states, streams, objects []string
for _, s := range m.State {
states = append(states, s.Name)
}
for _, s := range m.Streams {
streams = append(streams, s.Name)
}
for _, o := range m.Objects {
objects = append(objects, o.Name)
}
var program catalogue.Artifact
if m.Build == nil {
t.Fatal("the manifest builds nothing")
}
for _, a := range m.Build.Artifacts {
if a.Name == "program" {
program = a
}
}
rows := []struct {
name string
ok bool
}{
// RN1/AC1 (§21): a foundation part asks no template; its kept data are in old sections.
{"no requires", len(m.Requires) == 0},
{"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat &&
equalSets(m.Claims[0].Serves, seat.Verbs())},
// §27, §85, §330, §355, §77: the old state sections.
{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "findings", "first-views"})},
// §27, §85: the step log on an old stream, its outputs in an old objects section.
{"old streams section", equalSets(streams, []string{seat.StateStepLog})},
{"step-log 128 MiB", len(m.Streams) == 1 && m.Streams[0].MaxMiB == 128},
{"old objects section", equalSets(objects, []string{seat.StateStepOutputs})},
{"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024},
// §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears.
{"consumes", equalSets(m.Consumes, []string{
"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed",
"operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended",
"node-build-image.step-ended", "mesh-wants.step-ended", "mesh-catalog.registration-changed",
"mesh-catalog.catalog-revision"})},
// RN1/AC5 (§95, §130): the new assigner is delivered before the runner.
{"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})},
}
for _, r := range rows {
if !r.ok {
t.Errorf("%s: not as the design says", r.name)
}
}
}
func equalSets(a, b []string) bool {
x, y := slices.Clone(a), slices.Clone(b)
slices.Sort(x)
slices.Sort(y)
return slices.Equal(x, y)
}
modules/mesh-delivery-pipeline/module.json
+100
View File
@@ -0,0 +1,100 @@
{
"module": "mesh-delivery-pipeline",
"version": "1",
"slug": "pipeline",
"claims": [
{
"name": "mesh-pipeline-runner",
"scope": "mesh",
"serves": [
"run",
"start",
"show",
"list",
"stands",
"why-carried",
"kept",
"review",
"rerun",
"accept",
"cancel",
"ask-again"
]
}
],
"consumes": [
"mesh-delivery-indexer.take-in.taken",
"mesh-builds.package-published",
"mesh-builds.toolchain-agreed",
"operator-channel.decided",
"node-build-go.step-ended",
"node-build-node.step-ended",
"node-build-image.step-ended",
"mesh-wants.step-ended",
"mesh-catalog.registration-changed",
"mesh-catalog.catalog-revision"
],
"state": [
{ "name": "pipelines", "history": 1 },
{ "name": "kept", "history": 1 },
{ "name": "aged", "history": 1 },
{ "name": "findings", "history": 1 },
{ "name": "first-views", "history": 1 }
],
"streams": [
{ "name": "step-log", "events": ["step-log.**"], "max-mib": 128 }
],
"objects": [
{ "name": "step-outputs", "max-mib": 1024 }
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
"secrets-owner": "mesh-delivery-pipeline",
"resources": [
{
"id": "account",
"type": "user",
"name": "mesh-delivery-pipeline",
"shell": "/usr/bin/nologin",
"home": "/var/lib/mesh-delivery-pipeline"
},
{
"id": "state",
"type": "directory",
"mode": "0700",
"owner": "mesh-delivery-pipeline",
"place": ".",
"unknown": "program"
},
{
"id": "pipeline",
"type": "process",
"name": "mesh-delivery-pipeline",
"artifact": "program",
"run": [
"./mesh-delivery-pipeline"
],
"env": {
"MESH_BROKER_FILE": "${dir:state}/broker"
},
"user": "mesh-delivery-pipeline",
"health": {
"kind": "unit"
}
}
],
"build": {
"artifacts": [
{
"name": "program",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-delivery-pipeline",
"binary": "mesh-delivery-pipeline",
"needs": ["mesh-delivery-assigner:program"]
}
]
}
}