From cafff10266988eb0d1e3bb7e0f7f0978be9957e9 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:32:31 +0100 Subject: [PATCH] RN1: the pipeline runner's module, a foundation part with its kept data in old sections (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-delivery-pipeline/main.go | 51 +++++++++ modules/mesh-delivery-pipeline/go.mod | 17 +++ modules/mesh-delivery-pipeline/go.sum | 12 +++ .../mesh-delivery-pipeline/manifest_test.go | 77 ++++++++++++++ modules/mesh-delivery-pipeline/module.json | 100 ++++++++++++++++++ 5 files changed, 257 insertions(+) create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go create mode 100644 modules/mesh-delivery-pipeline/go.mod create mode 100644 modules/mesh-delivery-pipeline/go.sum create mode 100644 modules/mesh-delivery-pipeline/manifest_test.go create mode 100644 modules/mesh-delivery-pipeline/module.json diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go new file mode 100644 index 00000000..a931e9eb --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -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, `_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, `_`. +func ownBucket(module, state string) string { return module + "_" + state } diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod new file mode 100644 index 00000000..511f96a5 --- /dev/null +++ b/modules/mesh-delivery-pipeline/go.mod @@ -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 +) diff --git a/modules/mesh-delivery-pipeline/go.sum b/modules/mesh-delivery-pipeline/go.sum new file mode 100644 index 00000000..0daef11f --- /dev/null +++ b/modules/mesh-delivery-pipeline/go.sum @@ -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= diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go new file mode 100644 index 00000000..59aa7378 --- /dev/null +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -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) +} diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json new file mode 100644 index 00000000..02802158 --- /dev/null +++ b/modules/mesh-delivery-pipeline/module.json @@ -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"] + } + ] + } +}