Merge pull request 'R4g (mesh-tools): seat.pull and seat.settle for the build modules, merge-check.sh and the root image module go (design 47)' (#40) from feat/r4g into main
Reviewed-on: #40
This commit was merged in pull request #40.
This commit is contained in:
-85
@@ -1,85 +0,0 @@
|
||||
ARG NODE_BASE=node:22-bookworm-slim
|
||||
# The mesh-tools module: the two images every TypeScript module is COMPILED in and may RUN in. The
|
||||
# runtime itself ships as the node-tools module's bundle (node-tools/, novox/hq ADR 0175, to-be 38
|
||||
# WP3); these images are the toolchain for TypeScript bundles and the base a module's own service
|
||||
# may still be built on. They are no longer how tools reach a node.
|
||||
#
|
||||
# **They were the same image, and that was a mistake.** A module's recipe starts from this and
|
||||
# invokes the compiler out of it, so the compiler had to be here — and because the same image was
|
||||
# also what every module ran in, every running container on every machine carried a TypeScript
|
||||
# compiler it would never invoke. The answer is separate stages rather than one image bad at both
|
||||
# jobs. `build.artifacts` in module.json names the published stage each — `toolchain` and `runtime`.
|
||||
#
|
||||
# The `deps` stage is neither published nor named there: it is where the mesh's package registry is
|
||||
# reached, so it is where — and only where — the credential to reach it exists. The toolchain copies
|
||||
# resolved node_modules out of it, so the credential is in no image any machine ever holds
|
||||
# (novox/hq ADR 0076). This is the buildkit-secret's job done without buildkit, because a machine's
|
||||
# docker may carry no buildx.
|
||||
|
||||
# ---- deps: node_modules resolved from the mesh's registry, credential and all ----------------
|
||||
FROM ${NODE_BASE} AS deps
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends git ca-certificates \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
WORKDIR /app
|
||||
COPY node-tools/package.json ./
|
||||
# The builder writes .npmrc into the build context; it authenticates to the mesh's package registry
|
||||
# for the @novox scope, which is where @novox/mesh-sdk resolves. This stage is not published, so the
|
||||
# credential travels no further than here. Development dependencies included: the compiler is one.
|
||||
COPY .npmrc ./.npmrc
|
||||
RUN npm install --no-audit --no-fund
|
||||
# **The SDK this image carries is the one the mesh last published** (novox/hq issue 212). The range in
|
||||
# package.json is resolved once and the layer above is cached, so a release reached no toolchain until
|
||||
# that file changed. The exact version arrives as a build argument from the SDK module's published
|
||||
# package (module.json `build.on`), so a new release is a new argument, this layer runs again — and
|
||||
# the planner orders this module after the SDK, so a release rebuilds the toolchain and, after it,
|
||||
# every bundle compiled in it (issue 211).
|
||||
ARG MESH_SDK=@novox/mesh-sdk@latest
|
||||
RUN npm install --no-audit --no-fund "${MESH_SDK}"
|
||||
|
||||
# ---- compiling: the runtime's own code built, WITHOUT the credential -------------------------
|
||||
FROM ${NODE_BASE} AS compiling
|
||||
WORKDIR /app
|
||||
COPY node-tools/package.json ./
|
||||
# The resolved libraries, but not the .npmrc that resolved them.
|
||||
COPY --from=deps /app/node_modules ./node_modules
|
||||
COPY node-tools/tsconfig.json ./
|
||||
COPY node-tools/src ./src
|
||||
RUN npm run build
|
||||
|
||||
# ---- what a running bundle needs, and nothing else -------------------------------------------
|
||||
# Its own stage so the toolchain image keeps its build tools while the runtime image does not.
|
||||
FROM compiling AS lean
|
||||
RUN npm prune --omit=dev
|
||||
|
||||
# ---- toolchain: what a TypeScript bundle is compiled in ---------------------------------------
|
||||
# The compiler, and esbuild (a development dependency) at /app/node_modules/esbuild: the builder
|
||||
# bundles every entrypoint and launcher into one file with it (novox/hq ADR 0193), so a bundle
|
||||
# carries what it imports and not this image's node_modules.
|
||||
# Beside the compiler, at /app/runtime, what every TypeScript bundle runs with: the production
|
||||
# dependencies the SDK and the runtime need, and a package.json saying the compiled files are ES
|
||||
# modules. The builder copies this directory whole into a compiled bundle (novox/hq ADR 0188 §5),
|
||||
# so a bundle unpacked on a machine starts — a `.js` without that package.json is read as
|
||||
# CommonJS, and an import of `nats` without node_modules beside it resolves to nothing.
|
||||
#
|
||||
# **The client libraries a catalogue module depends on are here too** (novox/hq issue 385): pg, mongodb and
|
||||
# mssql, development dependencies of node-tools, so the runtime stage prunes them. A catalogue's
|
||||
# merge-check-typescript.sh fetches nothing and copies this image's node_modules beside a module; a module
|
||||
# depending on a package that is not here cannot be type-checked or tested, and fails its check.
|
||||
FROM compiling AS toolchain
|
||||
COPY --from=lean /app/node_modules /app/runtime/node_modules
|
||||
RUN printf '{"type":"module","private":true}\n' > /app/runtime/package.json
|
||||
# **git, because a pull request's merge check runs in this image too** (novox/hq ADR 0237): a repository's
|
||||
# own merge-check.sh that declares this toolchain runs here, and a suite that reads a repository's history
|
||||
# (the lab's last-run record) ran into "spawnSync git ENOENT". Nothing compiled here calls it.
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends git \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
# ---- runtime: what a module's own service may run in ------------------------------------------
|
||||
FROM ${NODE_BASE} AS runtime
|
||||
WORKDIR /app
|
||||
COPY node-tools/package.json ./
|
||||
COPY --from=lean /app/node_modules ./node_modules
|
||||
COPY --from=compiling /app/dist ./dist
|
||||
ENTRYPOINT ["node", "dist/main.js"]
|
||||
@@ -1,13 +1,11 @@
|
||||
# mesh-tools
|
||||
|
||||
Two modules in one repository (novox/hq ADR 0069), one piece of software:
|
||||
One module, `node-tools`; the toolchain image module that stood beside it is retired (novox/hq ADR 0524 §5):
|
||||
|
||||
- **`node-tools`** (`node-tools/`) — the node's **tool runtime** as a module (ADR 0175, to-be 38
|
||||
WP3): one process per machine the host runs from this bundle, serving every assigned module's tools
|
||||
and every held seat's verbs on the bus, and answering MCP on the machine's loopback — the console
|
||||
(design 34). The code, its tests and the `mesh` client all live there.
|
||||
- **`mesh-tools`** (this directory) — the two images TypeScript bundles are compiled in and a module's
|
||||
own *service* may still run in. Built from the same code; no longer how tools reach a node.
|
||||
|
||||
The runtime:
|
||||
|
||||
@@ -49,8 +47,7 @@ MESH_BROKER_URL a plain URL instead of the credential, for the bootstrap
|
||||
|
||||
`node dist/main.js`. On a node the controller composes the variables and the host supervises the
|
||||
process like any other host-side workload (novox/hq to-be 38). As `node-tools` the same process is
|
||||
the console: MCP on `127.0.0.1:4270` (or `MESH_CONSOLE_LISTEN`). The container (`Dockerfile`) is how
|
||||
a module's own *service* may still be built; it is no longer how tools reach a node.
|
||||
the console: MCP on `127.0.0.1:4270` (or `MESH_CONSOLE_LISTEN`).
|
||||
|
||||
## `mesh` — the tools for whoever is on a machine
|
||||
|
||||
|
||||
@@ -1,56 +0,0 @@
|
||||
#!/bin/sh
|
||||
# mesh-check-toolchain: go
|
||||
#
|
||||
# The tool runtime's own check (novox/hq ADR 0237 as amended): the second layer of a pull request's merge
|
||||
# check, `mesh/repo-check`, run by the build seat in the mesh's Go toolchain — and by hand.
|
||||
#
|
||||
# The gate — every machine of the facts snapshot composed with this change — is the build seat's first
|
||||
# layer (`mesh/merge-gate`), run because the mesh's module graph builds node-tools and mesh-tools from
|
||||
# here. This is the code's own: node-tools (Go) formatted, vetted and its suite, every test on a bus of its
|
||||
# own at the release the mesh runs (internal/meshtest), packages in parallel, under the race detector when
|
||||
# the toolchain has a C compiler.
|
||||
#
|
||||
# **Said, never passed silently**: the runtime's and the console's tests launch TypeScript and Python
|
||||
# bundles that import the mesh's own package (@novox/mesh-sdk), which comes from the package registry — and
|
||||
# a check of code nobody has approved is given no credential for it. Where node and that package are not
|
||||
# present those two packages are not tested here, and the TypeScript in node-tools/src neither.
|
||||
set -eu
|
||||
cd node-tools
|
||||
|
||||
unformatted=$(gofmt -l cmd internal)
|
||||
if [ -n "$unformatted" ]; then
|
||||
echo "not gofmt'd:"
|
||||
echo "$unformatted"
|
||||
exit 1
|
||||
fi
|
||||
CGO_ENABLED=0 go vet ./...
|
||||
|
||||
packages=$(go list ./...)
|
||||
bundleless=""
|
||||
if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; then
|
||||
echo "NOT TESTED HERE: internal/runtime and internal/console launch bundles that need node and @novox/mesh-sdk"
|
||||
packages=$(printf '%s\n' "$packages" | grep -v -e '/internal/runtime$' -e '/internal/console$')
|
||||
# What of the console launches no bundle runs anyway: how an address resolves (novox/hq issue 287),
|
||||
# and how a search ranks (novox/hq ADR 0245), and what discovery heard and how a round that heard nothing is
|
||||
# answered (novox/hq issue 554), and how a session waits on the bus (novox/hq ADR 0367).
|
||||
bundleless='^(TestMeshWait.*|TestTheManifestNamesTheFirstFamiliesOneByOne|TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces|TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed|TestARuntimeHeardBeforeAndNotNowIsNamed|TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore|TestAnAddressKnownOnlyFromAnEarlierAnswerSaysHowOld)$'
|
||||
# And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its
|
||||
# environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write
|
||||
# by compare-and-set loses no update under contention (novox/hq research 042), and a module's log verbs
|
||||
# (novox/hq ADR 0297), and a key's history (novox/hq ADR 0307), and a module's values asked of its machine's
|
||||
# provisioner client (novox/hq ADR 0370).
|
||||
runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestAKeysHistoryIsEveryValueKeptOldestFirst|TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState|TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone|TestABundleAsksForItsValuesAsItsOwnModule|TestABundlesWatchIsHandedTheValueAndEachChange)$'
|
||||
fi
|
||||
if command -v gcc >/dev/null 2>&1; then
|
||||
CGO_ENABLED=1 go test -race -count=1 $packages
|
||||
else
|
||||
echo "NOT RACE-CHECKED: the toolchain holds no C compiler; the suite runs without the race detector"
|
||||
CGO_ENABLED=0 go test -count=1 $packages
|
||||
fi
|
||||
if [ -n "$bundleless" ]; then
|
||||
CGO_ENABLED=0 go test -count=1 -run "$bundleless" ./internal/console
|
||||
fi
|
||||
if [ -n "${runtimeless:-}" ]; then
|
||||
CGO_ENABLED=0 go test -count=1 -run "$runtimeless" ./internal/runtime
|
||||
fi
|
||||
echo "NOT TESTED HERE: node-tools/src (TypeScript) needs @novox/mesh-sdk from the package registry"
|
||||
-33
@@ -1,33 +0,0 @@
|
||||
{
|
||||
"module": "mesh-tools",
|
||||
"version": "1",
|
||||
"slug": "tools",
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "build",
|
||||
"kind": "image",
|
||||
"from": "Dockerfile",
|
||||
"target": "toolchain"
|
||||
},
|
||||
{
|
||||
"name": "runtime",
|
||||
"kind": "image",
|
||||
"from": "Dockerfile",
|
||||
"target": "runtime"
|
||||
}
|
||||
],
|
||||
"on": [
|
||||
{
|
||||
"arg": "NODE_BASE",
|
||||
"image": "node@sha256:48e4b67d85f87bd551df43704e24d252f56cc5f8e9718841aace50f19948f0f9"
|
||||
},
|
||||
{
|
||||
"arg": "MESH_SDK",
|
||||
"module": "mesh-sdk",
|
||||
"artifact": "lib"
|
||||
}
|
||||
]
|
||||
},
|
||||
"resources": []
|
||||
}
|
||||
+1
-1
@@ -8,7 +8,7 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
|
||||
+2
-2
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 h1:mAMt3KFmrHzDiBv4MLL94opWcG9y73ISMrBUyisapQE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// **A holder that pulls one piece of work when it has room** (novox/hq ADR 0525 §4, §6; issue 1887 A11). SeatTake
|
||||
// hands a bundle every piece a worker holds, one at a time and prefetched; a build module must instead take one ask
|
||||
// only when it has a free slot, from its workers in their order (delivery, then check, then recheck), and decide each
|
||||
// ask's fate when its step ends. SeatPull is one fetch of at most one message from a worker the module's membership
|
||||
// lists, never a prefetch; SeatSettle gives a pulled message its fate. A message the bundle never settles (its process
|
||||
// died) is delivered again by the worker's own acknowledgement wait.
|
||||
|
||||
// MaxPullWait bounds how long one pull waits for a message: under the bundle's 30 s wait for its answer.
|
||||
var MaxPullWait = 25 * time.Second
|
||||
|
||||
// Pulled is one message a module pulled from a seat's worker, named by an id the runtime gave it.
|
||||
type Pulled struct {
|
||||
ID string `json:"id"`
|
||||
Worker string `json:"worker"`
|
||||
Subject string `json:"subject"`
|
||||
Body json.RawMessage `json:"body"`
|
||||
Headers map[string]string `json:"headers,omitempty"`
|
||||
Delivered uint64 `json:"delivered"`
|
||||
}
|
||||
|
||||
// pulls are the messages modules pulled and have not settled, by id; each kept with the module it was pulled for, so
|
||||
// one module never settles another's.
|
||||
type pulls struct {
|
||||
mu sync.Mutex
|
||||
held map[string]pulledMsg
|
||||
cons map[string]jetstream.Consumer // by module and consumer
|
||||
}
|
||||
|
||||
type pulledMsg struct {
|
||||
module string
|
||||
msg jetstream.Msg
|
||||
}
|
||||
|
||||
var seatPulls = &pulls{held: map[string]pulledMsg{}, cons: map[string]jetstream.Consumer{}}
|
||||
|
||||
func pullID() string {
|
||||
var b [8]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
|
||||
// SeatPull fetches at most one message from consumer for module, waiting at most wait (none when 0): nil when none
|
||||
// came. The consumer must be one the module's membership lists as its worker (novox/hq ADR 0259); the runtime binds to
|
||||
// it by name and never makes one.
|
||||
func (c *Conn) SeatPull(module, consumer string, wait time.Duration) (*Pulled, error) {
|
||||
t, err := c.trafficOf(module)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var w *SeatWorker
|
||||
var names []string
|
||||
for i := range t.Workers {
|
||||
names = append(names, t.Workers[i].Consumer)
|
||||
if t.Workers[i].Consumer == consumer {
|
||||
w = &t.Workers[i]
|
||||
}
|
||||
}
|
||||
if w == nil {
|
||||
return nil, fmt.Errorf("%s pulls no work from %s: its membership lists %s (novox/hq ADR 0259)", module,
|
||||
consumer, strings.Join(names, ", "))
|
||||
}
|
||||
if wait > MaxPullWait {
|
||||
wait = MaxPullWait
|
||||
}
|
||||
bound, err := c.pullConsumer(module, *w)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var batch jetstream.MessageBatch
|
||||
if wait <= 0 {
|
||||
batch, err = bound.FetchNoWait(1)
|
||||
} else {
|
||||
batch, err = bound.Fetch(1, jetstream.FetchMaxWait(wait))
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("pulling %s's worker %s: %w", module, w.Consumer, err)
|
||||
}
|
||||
msg, ok := <-batch.Messages()
|
||||
if !ok {
|
||||
if err := batch.Error(); err != nil {
|
||||
return nil, fmt.Errorf("pulling %s's worker %s: %w", module, w.Consumer, err)
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
if !json.Valid(msg.Data()) {
|
||||
_ = msg.Term()
|
||||
return nil, nil
|
||||
}
|
||||
headers := map[string]string{}
|
||||
for k := range msg.Headers() {
|
||||
headers[k] = msg.Headers().Get(k)
|
||||
}
|
||||
delivered := uint64(1)
|
||||
if meta, merr := msg.Metadata(); merr == nil && meta != nil {
|
||||
delivered = meta.NumDelivered
|
||||
}
|
||||
p := &Pulled{ID: pullID(), Worker: w.Consumer, Subject: msg.Subject(), Body: json.RawMessage(msg.Data()),
|
||||
Headers: headers, Delivered: delivered}
|
||||
seatPulls.mu.Lock()
|
||||
seatPulls.held[p.ID] = pulledMsg{module: module, msg: msg}
|
||||
seatPulls.mu.Unlock()
|
||||
return p, nil
|
||||
}
|
||||
|
||||
func (c *Conn) pullConsumer(module string, w SeatWorker) (jetstream.Consumer, error) {
|
||||
key := module + " " + w.Consumer
|
||||
seatPulls.mu.Lock()
|
||||
bound, ok := seatPulls.cons[key]
|
||||
seatPulls.mu.Unlock()
|
||||
if ok {
|
||||
return bound, nil
|
||||
}
|
||||
js, err := jetstream.New(c.nc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
bound, err = js.Consumer(ctx, w.Stream, w.Consumer)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("binding %s's worker %s on %s: %w", module, w.Consumer, w.Stream, err)
|
||||
}
|
||||
seatPulls.mu.Lock()
|
||||
seatPulls.cons[key] = bound
|
||||
seatPulls.mu.Unlock()
|
||||
return bound, nil
|
||||
}
|
||||
|
||||
// The fates SeatSettle gives a pulled message.
|
||||
const (
|
||||
SettleAck = "ack" // done: never delivered again
|
||||
SettleNak = "nak" // handed back, to be delivered again at once
|
||||
SettleTerm = "term" // dropped
|
||||
SettleProgress = "progress" // still being worked: its acknowledgement wait starts again
|
||||
)
|
||||
|
||||
// SeatSettle gives the message module pulled under id its fate. A message pulled for another module, or settled
|
||||
// already, is refused.
|
||||
func (c *Conn) SeatSettle(module, id, fate string) error {
|
||||
seatPulls.mu.Lock()
|
||||
p, ok := seatPulls.held[id]
|
||||
if ok && p.module != module {
|
||||
ok = false
|
||||
}
|
||||
if ok && fate != SettleProgress {
|
||||
delete(seatPulls.held, id)
|
||||
}
|
||||
seatPulls.mu.Unlock()
|
||||
if !ok {
|
||||
return fmt.Errorf("%s holds no pulled message %s", module, id)
|
||||
}
|
||||
switch fate {
|
||||
case SettleAck:
|
||||
return p.msg.Ack()
|
||||
case SettleNak:
|
||||
return p.msg.Nak()
|
||||
case SettleTerm:
|
||||
return p.msg.Term()
|
||||
case SettleProgress:
|
||||
return p.msg.InProgress()
|
||||
}
|
||||
seatPulls.mu.Lock()
|
||||
seatPulls.held[id] = p // an unknown fate settles nothing
|
||||
seatPulls.mu.Unlock()
|
||||
return fmt.Errorf("a pulled message's fate is ack, nak, term or progress, not %q", fate)
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package bus_test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
)
|
||||
|
||||
// 1887/A11: a holder pulls one message at a time, only from a worker its membership lists, never a prefetch; it settles
|
||||
// only what it pulled, and a message handed back comes again.
|
||||
func TestAHolderPullsOneMessageWhenItHasRoom(t *testing.T) {
|
||||
_, conn, js := seatMesh(t)
|
||||
for _, id := range []string{"a1", "a2"} {
|
||||
if _, err := conn.SeatPublish("mesh-delivery", "mesh.seat.operator-channel.accept.ask.mesh-delivery",
|
||||
json.RawMessage(`{"id":"`+id+`"}`), id); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
first, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0)
|
||||
if err != nil || first == nil || string(first.Body) != `{"id":"a1"}` || first.Delivered != 1 {
|
||||
t.Fatalf("the first pull = %+v, %v", first, err)
|
||||
}
|
||||
// One pulled, one still waiting on the stream: nothing was fetched ahead.
|
||||
info, err := js.ConsumerInfo("SEAT_OPERATOR_CHANNEL", "SEAT_OPERATOR_CHANNEL_worker")
|
||||
if err != nil || info.NumPending != 1 || info.NumAckPending != 1 {
|
||||
t.Fatalf("after one pull: pending %d, in hand %d (%v); want 1 and 1", info.NumPending, info.NumAckPending, err)
|
||||
}
|
||||
second, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0)
|
||||
if err != nil || second == nil || string(second.Body) != `{"id":"a2"}` {
|
||||
t.Fatalf("the second pull = %+v, %v", second, err)
|
||||
}
|
||||
if none, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0); err != nil || none != nil {
|
||||
t.Fatalf("an empty worker answered %+v, %v", none, err)
|
||||
}
|
||||
began := time.Now()
|
||||
if none, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 200*time.Millisecond); err != nil || none != nil || time.Since(began) < 150*time.Millisecond {
|
||||
t.Fatalf("a waiting pull answered %+v, %v after %v", none, err, time.Since(began))
|
||||
}
|
||||
if err := conn.SeatSettle("telegram", first.ID, bus.SettleAck); err == nil {
|
||||
t.Error("a module settled another's message")
|
||||
}
|
||||
if err := conn.SeatSettle("messenger", first.ID, bus.SettleProgress); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := conn.SeatSettle("messenger", first.ID, bus.SettleAck); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := conn.SeatSettle("messenger", first.ID, bus.SettleAck); err == nil {
|
||||
t.Error("a message was settled twice")
|
||||
}
|
||||
if err := conn.SeatSettle("messenger", second.ID, bus.SettleNak); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", time.Second)
|
||||
if err != nil || again == nil || string(again.Body) != `{"id":"a2"}` || again.Delivered != 2 {
|
||||
t.Fatalf("the message handed back = %+v, %v", again, err)
|
||||
}
|
||||
if _, err := conn.SeatPull("telegram", "SEAT_OPERATOR_CHANNEL_worker", 0); err == nil || !strings.Contains(err.Error(), "membership lists") {
|
||||
t.Errorf("a channel pulled the router's work: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -39,6 +39,7 @@ func call(t *testing.T, endpoint, tool string, args map[string]any) (string, boo
|
||||
// novox/hq ADR 0195 and ADR 0367: the console announces seven tools, and everything the mesh answers is reached
|
||||
// through them by one address per layer.
|
||||
func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
was := IndexKept
|
||||
IndexKept = 0 // every discovery asks the mesh, so a module arriving mid-test is found
|
||||
t.Cleanup(func() { IndexKept = was })
|
||||
|
||||
@@ -42,6 +42,7 @@ func post(t *testing.T, endpoint string, body string) map[string]any {
|
||||
// As node-tools, the runtime serves the bundles and is the console on loopback: the listing is what
|
||||
// the modules and the mesh's records answered, and a call reaches the module on the machine named.
|
||||
func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
mesh.Issue(t, mt.MembershipOf("alpha", "desk", true, nil))
|
||||
mesh.Issue(t, mt.MembershipOf("beta", "desk", false, map[string][]string{"node-shelf": {"list", "clear"}}))
|
||||
|
||||
@@ -288,7 +288,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
|
||||
subscriber = c
|
||||
mu.Unlock()
|
||||
err = mesh.Subscribe(deliver)
|
||||
case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds":
|
||||
case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds", "mesh/seat.pull", "mesh/seat.settle":
|
||||
var answered json.RawMessage
|
||||
if answered, err = mesh.Seat(strings.TrimPrefix(m.Method, "mesh/seat."), m.Params, nil); err == nil {
|
||||
result = answered
|
||||
|
||||
@@ -69,3 +69,45 @@ func TestABundleTakesASeatsWorkThroughItsOwnChannel(t *testing.T) {
|
||||
t.Fatalf("the work was not handed to the bundle: %s %v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A bundle that pulls one ask from its worker when it is listed.
|
||||
const pullingBundle = `#!/bin/sh
|
||||
while IFS= read -r line; do
|
||||
id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p')
|
||||
case "$line" in
|
||||
*'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
|
||||
*'"method":"tools/list"'*)
|
||||
printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id"
|
||||
printf '{"jsonrpc":"2.0","id":"p1","method":"mesh/seat.pull","params":{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":0}}\n' ;;
|
||||
esac
|
||||
done
|
||||
`
|
||||
|
||||
// 1887/A11: a bundle's pull reaches the runtime's bus from its own channel, as its other seat traffic does.
|
||||
func TestABundlePullsASeatsWorkThroughItsOwnChannel(t *testing.T) {
|
||||
entry := filepath.Join(t.TempDir(), "bundle")
|
||||
if err := os.WriteFile(entry, []byte(pullingBundle), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b := &seating{}
|
||||
l, err := Start("mesh-build-go", entry, os.Environ(), b, t.Logf)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(l.Stop)
|
||||
for i := 0; ; i++ {
|
||||
b.mu.Lock()
|
||||
verb, params := b.verb, b.params
|
||||
b.mu.Unlock()
|
||||
if verb == "pull" {
|
||||
if params != `{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":0}` {
|
||||
t.Fatalf("pulled with %s", params)
|
||||
}
|
||||
return
|
||||
}
|
||||
if i > 100 {
|
||||
t.Fatal("the bundle's pull never reached the runtime")
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
package meshtest
|
||||
|
||||
import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// bundlesWhy returns the reason bundles cannot be launched from a checkout rooted at root, or "" when they can: the
|
||||
// TypeScript and Python bundles the runtime and the console launch need node and the package
|
||||
// @novox/mesh-sdk, which comes from the package registry and is installed beside node-tools/package.json
|
||||
// (novox/hq ADR 0529 clause 1: what merge-check.sh said "NOT TESTED HERE" about is a skip with its reason).
|
||||
func bundlesWhy(root string, haveNode bool) string {
|
||||
if !haveNode {
|
||||
return "node is not installed"
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "node_modules", "@novox", "mesh-sdk")); err != nil {
|
||||
return "@novox/mesh-sdk is not installed in " + filepath.Join(root, "node_modules")
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// NeedBundles skips the test, saying why, where the bundles it launches cannot run here.
|
||||
func NeedBundles(t *testing.T) {
|
||||
t.Helper()
|
||||
_, err := exec.LookPath("node")
|
||||
if why := bundlesWhy(moduleRoot(), err == nil); why != "" {
|
||||
t.Skip("NOT TESTED HERE: this test launches bundles, and " + why)
|
||||
}
|
||||
}
|
||||
|
||||
// moduleRoot is node-tools/, where package.json and node_modules are.
|
||||
func moduleRoot() string {
|
||||
_, here, _, _ := runtime.Caller(0)
|
||||
return filepath.Join(filepath.Dir(here), "..", "..")
|
||||
}
|
||||
@@ -73,3 +73,49 @@ func TestEachTestHasABusOfItsOwn(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// What merge-check.sh said was "NOT TESTED HERE" is a skip with its reason: no node, or no package, each says so.
|
||||
func TestBundlesCannotBeLaunchedWithoutNodeOrTheSDKAndTheSkipSaysWhy(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
if why := bundlesWhy(root, false); !strings.Contains(why, "node is not installed") {
|
||||
t.Errorf("no node said %q", why)
|
||||
}
|
||||
if why := bundlesWhy(root, true); !strings.Contains(why, "@novox/mesh-sdk") {
|
||||
t.Errorf("no package said %q", why)
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Join(root, "node_modules", "@novox", "mesh-sdk"), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if why := bundlesWhy(root, true); why != "" {
|
||||
t.Errorf("node and the package present said %q", why)
|
||||
}
|
||||
}
|
||||
|
||||
// The checker runs node-tools' Go level by Go's convention and its Node level by package.json's test script
|
||||
// (novox/hq ADR 0529): no script of the repository's own runs beside them, and no module but node-tools is built from
|
||||
// here (ADR 0524 §5, the toolchain image retired).
|
||||
func TestTheRepositoryHoldsNoCheckScriptAndNodeToolsIsItsOnlyModule(t *testing.T) {
|
||||
root := filepath.Join("..", "..", "..")
|
||||
if _, err := os.Stat(filepath.Join(root, "merge-check.sh")); err == nil {
|
||||
t.Errorf("merge-check.sh is back: the checker's levels run the tests")
|
||||
}
|
||||
var found []string
|
||||
err := filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if d.IsDir() && (d.Name() == "node_modules" || d.Name() == ".git") {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
if !d.IsDir() && d.Name() == "module.json" {
|
||||
found = append(found, filepath.ToSlash(p))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(found) != 1 || !strings.HasSuffix(found[0], "/node-tools/module.json") {
|
||||
t.Errorf("the module manifests are %v, want node-tools' alone", found)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
// read here with NATS's own types, each endpoint a subject actually served — and the announcement
|
||||
// follows a re-issued membership.
|
||||
func TestTheRuntimeAnnouncesWhatItServesInTheServicesProtocol(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
mesh.Issue(t, mt.MembershipOf("alpha", "anchor", false, nil))
|
||||
mesh.Issue(t, mt.MembershipOf("beta", "anchor", false, map[string][]string{"node-shelf": {"list", "clear"}}))
|
||||
|
||||
@@ -50,6 +50,7 @@ func countLines(t *testing.T, path, want string) int {
|
||||
// is its bus — it binds the module's own consumer, hands each event to the bundle, and acknowledges it
|
||||
// only when the bundle has taken it; one the bundle failed or died on is offered again.
|
||||
func TestTheRuntimeHandsAModulesEventsToItsBundleAndAcknowledgesThemOnlyWhenTaken(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
restore := bus.NakDelay
|
||||
bus.NakDelay = 300 * time.Millisecond
|
||||
@@ -128,6 +129,7 @@ func TestTheRuntimeHandsAModulesEventsToItsBundleAndAcknowledgesThemOnlyWhenTake
|
||||
|
||||
// A long-running bundle that exits is started again, without waiting for a call (ADR 0198).
|
||||
func TestALongRunningBundleThatExitsIsStartedAgain(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
firstPause := launch.RestartFirst
|
||||
launch.RestartFirst = 100 * time.Millisecond
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/announce"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
@@ -759,6 +760,9 @@ type seatAsked struct {
|
||||
Worker string `json:"worker"`
|
||||
Bucket string `json:"bucket"`
|
||||
Key string `json:"key"`
|
||||
// WaitMS is how long a pull waits for a message (none when 0); Fate what a settle gives one (issue 1887 A11).
|
||||
WaitMS int `json:"wait_ms"`
|
||||
Fate string `json:"fate"`
|
||||
}
|
||||
|
||||
// Seat answers a bundle's seat traffic, each as the module and only as far as its membership lists:
|
||||
@@ -792,6 +796,19 @@ func (b *moduleBus) Seat(verb string, params json.RawMessage, handOn func(string
|
||||
return json.RawMessage("null"), err
|
||||
}
|
||||
return json.Marshal(entry)
|
||||
case "pull":
|
||||
// One message from a worker the membership lists, never a prefetch: a build module pulls an ask only when it
|
||||
// has room, from its workers in their order (novox/hq ADR 0525 §4; issue 1887 A11).
|
||||
p, err := conn.SeatPull(b.module, asked.Worker, time.Duration(asked.WaitMS)*time.Millisecond)
|
||||
if err != nil || p == nil {
|
||||
return json.RawMessage("null"), err
|
||||
}
|
||||
return json.Marshal(p)
|
||||
case "settle":
|
||||
if err := conn.SeatSettle(b.module, asked.ID, asked.Fate); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.RawMessage(`{}`), nil
|
||||
case "take":
|
||||
key := "take " + asked.Worker
|
||||
if b.bound(key) {
|
||||
|
||||
@@ -46,6 +46,7 @@ func same(t *testing.T, got, want string) {
|
||||
// The node's runtime serves five modules' bundles on one credential — two TypeScript, one broken,
|
||||
// one Python, one written against the protocol — and follows a re-issued membership (ADR 0175, 0193).
|
||||
func TestTheNodesRuntimeServesFiveModulesAndFollowsAReissuedMembership(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
mesh.Issue(t, mt.MembershipOf("alpha", "anchor", true, nil))
|
||||
mesh.Issue(t, mt.MembershipOf("beta", "anchor", false, map[string][]string{"node-shelf": {"list", "clear"}}))
|
||||
@@ -152,6 +153,7 @@ func TestTheNodesRuntimeServesFiveModulesAndFollowsAReissuedMembership(t *testin
|
||||
// Each bundle is given its own environment and none of another's (ADR 0192), and the composed
|
||||
// environments are not left in the runtime's.
|
||||
func TestEachBundleIsGivenItsOwnEnvironment(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
for _, m := range []string{"gamma", "delta", "zeta"} {
|
||||
mesh.Issue(t, mt.MembershipOf(m, "anchor", false, nil))
|
||||
@@ -192,6 +194,7 @@ func TestEachBundleIsGivenItsOwnEnvironment(t *testing.T) {
|
||||
|
||||
// An entrypoint that is not executable is refused by name, and the others serve (ADR 0193).
|
||||
func TestAnEntrypointThatIsNotExecutableIsRefused(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
mesh.Issue(t, mt.MembershipOf("alpha", "anchor", false, nil))
|
||||
mesh.Issue(t, mt.MembershipOf("plain", "anchor", false, nil))
|
||||
@@ -218,6 +221,7 @@ func TestAnEntrypointThatIsNotExecutableIsRefused(t *testing.T) {
|
||||
|
||||
// A launched bundle is told the module it serves, so its seat's verbs stay the seat's (ADR 0193).
|
||||
func TestALaunchedBundleRegisteringItsSeatFirstServesTheSeat(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
mesh.Issue(t, mt.MembershipOf("theta", "anchor", false, map[string][]string{"node-shelf": {"list"}}))
|
||||
nodeTools := connect(t, "node-tools", "anchor")
|
||||
|
||||
@@ -147,3 +147,65 @@ func TestALostCompareAndSetCarriesItsCode(t *testing.T) {
|
||||
t.Error("an update of revision zero was taken")
|
||||
}
|
||||
}
|
||||
|
||||
// 1887/A11: a bundle pulls one message from its own worker and settles it through its module's bus; another module may
|
||||
// neither pull that worker nor settle what it did not pull.
|
||||
func TestABundlePullsAndSettlesThroughItsModulesBus(t *testing.T) {
|
||||
mesh := mt.New(t)
|
||||
nc, err := nats.Connect(mt.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, _ := nc.JetStream()
|
||||
if _, err := js.AddStream(&nats.StreamConfig{Name: "SEAT_NODE_BUILD_GO", Retention: nats.WorkQueuePolicy,
|
||||
Subjects: []string{"mesh.seat.node-build-go.accept.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.AddConsumer("SEAT_NODE_BUILD_GO", &nats.ConsumerConfig{Durable: "SEAT_NODE_BUILD_GO_DELIVERY_worker",
|
||||
AckPolicy: nats.AckExplicitPolicy, FilterSubject: "mesh.seat.node-build-go.accept.delivery"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.Publish("mesh.seat.node-build-go.accept.delivery", []byte(`{"id":"ask1"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
holder := mt.MembershipOf("mesh-build-go", "anchor", false, nil)
|
||||
holder.SeatTraffic = &bus.SeatTraffic{Workers: []bus.SeatWorker{{Stream: "SEAT_NODE_BUILD_GO",
|
||||
Consumer: "SEAT_NODE_BUILD_GO_DELIVERY_worker", Filter: "mesh.seat.node-build-go.accept.delivery"}}}
|
||||
other := mt.MembershipOf("mesh-build-node", "anchor", false, nil)
|
||||
other.SeatTraffic = &bus.SeatTraffic{}
|
||||
mesh.Issue(t, holder)
|
||||
mesh.Issue(t, other)
|
||||
conn := connect(t, "node-tools", "anchor")
|
||||
conn.Follow("mesh-build-go")
|
||||
conn.Follow("mesh-build-node")
|
||||
mt.Until(t, func() error {
|
||||
if conn.Membership("mesh-build-go") == nil || conn.Membership("mesh-build-node") == nil {
|
||||
return errors.New("not issued yet")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
|
||||
building := all.forModule("mesh-build-go").(*moduleBus)
|
||||
elsewhere := all.forModule("mesh-build-node").(*moduleBus)
|
||||
if _, err := elsewhere.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker"}`), nil); err == nil {
|
||||
t.Fatal("another module pulled the Go build seat's worker")
|
||||
}
|
||||
raw, err := building.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":1000}`), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var p bus.Pulled
|
||||
if err := json.Unmarshal(raw, &p); err != nil || string(p.Body) != `{"id":"ask1"}` || p.ID == "" {
|
||||
t.Fatalf("pulled %s (%v)", raw, err)
|
||||
}
|
||||
if _, err := elsewhere.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err == nil {
|
||||
t.Fatal("another module settled the pulled ask")
|
||||
}
|
||||
if _, err := building.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if none, err := building.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker"}`), nil); err != nil || string(none) != "null" {
|
||||
t.Fatalf("an empty worker answered %s, %v", none, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -139,6 +139,7 @@ func waitFor(t *testing.T, path, line string) {
|
||||
// The TypeScript SDK against the runtime (ADR 0201): two watches of one state, each handed only its
|
||||
// own changes, the current values handled before the module's import goes on.
|
||||
func TestTheSDKsStateReachesTheRuntime(t *testing.T) {
|
||||
mt.NeedBundles(t)
|
||||
mesh := mt.New(t)
|
||||
kv := mesh.Bucket(t, "sdkkeeper_servers")
|
||||
if _, err := kv.Put("all.one", []byte(`{"url":"http://one"}`)); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user