Bus state by template: the typed requires entry, the binding and the calls package (issue 1966, B1) #164
@@ -0,0 +1,382 @@
|
||||
// Package buscalls is the contract of the calls a bundle makes on its own bus state through the tool runner, over the
|
||||
// bundle's pipe (novox/hq ADR 0595 §9, ADR 0597 §10, design 25 § 12.5): the object store's, a stream's read-back and
|
||||
// fill, a work queue's size and dead letters, and the ledger's. Each call names an entry by the local name its module
|
||||
// asks for it by in `requires`; the tool runner serves it from the module's membership's bindings, its own entries
|
||||
// only. Only the shapes and the method names are here: what the tool runner refuses, and why, is its own.
|
||||
//
|
||||
// It is a package of its own, imported only by the modules that use these calls and by the tool runner, so a change
|
||||
// here rebuilds those and not the module library's importers (ADR 0595 §9). It imports nothing of mesh-sdk.
|
||||
package buscalls
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MetaCall is where in a `tools/call`'s `_meta` the tool runner hands a bundle the id of the tool call in flight: the id
|
||||
// ApplyRequest.Call names, so the runner stamps the caller it proved for that call as the operation's author (novox/hq
|
||||
// ADR 0597 §7). The runner writes `_meta`; a caller writes only `arguments`, so no caller can name another call.
|
||||
const MetaCall = "mesh/call"
|
||||
|
||||
// The methods a bundle asks the tool runner, one per call.
|
||||
const (
|
||||
ObjectsPut = "mesh/objects.put"
|
||||
ObjectsGet = "mesh/objects.get"
|
||||
ObjectsList = "mesh/objects.list"
|
||||
ObjectsDelete = "mesh/objects.delete"
|
||||
ObjectsInfo = "mesh/objects.info"
|
||||
ObjectsWatch = "mesh/objects.watch"
|
||||
|
||||
StreamRead = "mesh/stream.read"
|
||||
StreamFill = "mesh/stream.fill"
|
||||
|
||||
WorkQueueSize = "mesh/workqueue.size"
|
||||
WorkQueueDead = "mesh/workqueue.dead"
|
||||
|
||||
LedgerApply = "mesh/ledger.apply"
|
||||
LedgerGet = "mesh/ledger.get"
|
||||
LedgerHistory = "mesh/ledger.history"
|
||||
LedgerWatch = "mesh/ledger.watch"
|
||||
LedgerQuery = "mesh/ledger.query"
|
||||
// The ledger's repairs, which the tool runner also serves as tools of every module owning a ledger, its own
|
||||
// ledgers only (design 25 § 12.5).
|
||||
LedgerRebuild = "mesh/ledger.rebuild"
|
||||
LedgerReindex = "mesh/ledger.reindex"
|
||||
LedgerSnapshot = "mesh/ledger.snapshot"
|
||||
LedgerLag = "mesh/ledger.lag"
|
||||
)
|
||||
|
||||
// The methods the tool runner sends the bundle for a watch it asked, each naming the watch by its id, as `mesh/log`
|
||||
// does for a log.
|
||||
const (
|
||||
ObjectsWatched = "mesh/objects"
|
||||
LedgerWatched = "mesh/ledger"
|
||||
)
|
||||
|
||||
// Fill is how much an entry holds against its cap: its messages (or objects), its bytes and its cap in bytes.
|
||||
type Fill struct {
|
||||
Count uint64 `json:"count"`
|
||||
Bytes uint64 `json:"bytes"`
|
||||
MaxBytes uint64 `json:"max_bytes"`
|
||||
}
|
||||
|
||||
// Done is the answer of a call that answers nothing but that it was done.
|
||||
type Done struct{}
|
||||
|
||||
// --- the object store
|
||||
|
||||
// ObjectsPutRequest stores one object whole under its name, replacing one of that name.
|
||||
type ObjectsPutRequest struct {
|
||||
Objects string `json:"objects"`
|
||||
Name string `json:"name"`
|
||||
Data []byte `json:"data"`
|
||||
Description string `json:"description,omitempty"`
|
||||
}
|
||||
|
||||
// ObjectsNameRequest names one object of an object store: get, delete.
|
||||
type ObjectsNameRequest struct {
|
||||
Objects string `json:"objects"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
// ObjectsRequest names an object store: list, info.
|
||||
type ObjectsRequest struct {
|
||||
Objects string `json:"objects"`
|
||||
}
|
||||
|
||||
// ObjectsWatchRequest asks the tool runner to send every object stored or deleted, as ObjectsChange under the watch's id.
|
||||
type ObjectsWatchRequest struct {
|
||||
Objects string `json:"objects"`
|
||||
Watch string `json:"watch"`
|
||||
}
|
||||
|
||||
// ObjectInfo is one object as its store holds it: put's answer, and each of list's.
|
||||
type ObjectInfo struct {
|
||||
Name string `json:"name"`
|
||||
Size uint64 `json:"size"`
|
||||
Chunks uint32 `json:"chunks"`
|
||||
Digest string `json:"digest"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Modified time.Time `json:"modified"`
|
||||
Deleted bool `json:"deleted,omitempty"`
|
||||
}
|
||||
|
||||
// ObjectData is get's answer: the object's info and its bytes.
|
||||
type ObjectData struct {
|
||||
Info ObjectInfo `json:"info"`
|
||||
Data []byte `json:"data"`
|
||||
}
|
||||
|
||||
// ObjectsListed is list's answer: every object the store holds.
|
||||
type ObjectsListed struct {
|
||||
Objects []ObjectInfo `json:"objects"`
|
||||
}
|
||||
|
||||
// ObjectsChange is one object stored or deleted, sent under ObjectsWatched; Current says it was there when the watch
|
||||
// began.
|
||||
type ObjectsChange struct {
|
||||
Watch string `json:"watch"`
|
||||
Object ObjectInfo `json:"object"`
|
||||
Current bool `json:"current"`
|
||||
}
|
||||
|
||||
// --- a stream
|
||||
|
||||
// StreamReadRequest reads a module's own stream from a sequence on, at most Limit messages, Subjects narrowing it.
|
||||
type StreamReadRequest struct {
|
||||
Stream string `json:"stream"`
|
||||
From uint64 `json:"from"`
|
||||
Subjects string `json:"subjects,omitempty"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
// StreamRequest names a stream: fill.
|
||||
type StreamRequest struct {
|
||||
Stream string `json:"stream"`
|
||||
}
|
||||
|
||||
// StreamMessage is one message of a stream.
|
||||
type StreamMessage struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Subject string `json:"subject"`
|
||||
Body json.RawMessage `json:"body"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// StreamPage is read's answer: the messages in order, the sequence to read on from (0 when nothing follows) and the
|
||||
// stream's last sequence when it was read.
|
||||
type StreamPage struct {
|
||||
Messages []StreamMessage `json:"messages"`
|
||||
Next uint64 `json:"next"`
|
||||
Last uint64 `json:"last"`
|
||||
}
|
||||
|
||||
// StreamFilled is fill's answer: the stream's fill and its first and last sequence.
|
||||
type StreamFilled struct {
|
||||
Fill
|
||||
First uint64 `json:"first"`
|
||||
Last uint64 `json:"last"`
|
||||
}
|
||||
|
||||
// --- a work queue
|
||||
|
||||
// WorkQueueRequest names a work queue: size.
|
||||
type WorkQueueRequest struct {
|
||||
Queue string `json:"queue"`
|
||||
}
|
||||
|
||||
// WorkQueueDeadRequest asks a work queue's dead letters, at most Limit of them.
|
||||
type WorkQueueDeadRequest struct {
|
||||
Queue string `json:"queue"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
// WorkQueueSized is size's answer: the messages waiting to be taken, against the queue's cap.
|
||||
type WorkQueueSized struct {
|
||||
Fill
|
||||
}
|
||||
|
||||
// DeadLetter is one message delivered as often as its queue allows and never settled.
|
||||
type DeadLetter struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Subject string `json:"subject"`
|
||||
Deliveries int `json:"deliveries"`
|
||||
Body json.RawMessage `json:"body,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// WorkQueueDeadLetters is dead's answer.
|
||||
type WorkQueueDeadLetters struct {
|
||||
Dead []DeadLetter `json:"dead"`
|
||||
}
|
||||
|
||||
// --- the ledger (ADR 0597)
|
||||
|
||||
// ApplyRequest applies one operation to one instance of a ledger. Call is the id of the tool call the operation
|
||||
// serves, as the tool runner handed it under MetaCall, whose caller the tool runner proved and stamps as the author
|
||||
// (ADR 0597 §7); empty, the operation carries the owning module's account alone. An author the bundle writes into Body
|
||||
// is never taken as the author.
|
||||
type ApplyRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Instance string `json:"instance"`
|
||||
Operation string `json:"operation"`
|
||||
Body json.RawMessage `json:"body,omitempty"`
|
||||
Call string `json:"call,omitempty"`
|
||||
}
|
||||
|
||||
// Applied is apply's answer: the operation was taken, with its sequence in the instance's history; StateBehind says
|
||||
// the state could not follow because its bucket is full, and catches up at the next apply after room is made (ADR
|
||||
// 0597 §6).
|
||||
type Applied struct {
|
||||
Taken bool `json:"taken"`
|
||||
Sequence uint64 `json:"sequence"`
|
||||
StateBehind bool `json:"state_behind,omitempty"`
|
||||
}
|
||||
|
||||
// Author is who an operation is by, as `apply` stamped it (ADR 0597 §7): the owning module's account, and the caller
|
||||
// of the tool call it served, as the tool runner proved it.
|
||||
type Author struct {
|
||||
Account string `json:"account"`
|
||||
Caller string `json:"caller,omitempty"`
|
||||
}
|
||||
|
||||
// InstanceRequest names one instance of a ledger: get, snapshot.
|
||||
type InstanceRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Instance string `json:"instance"`
|
||||
}
|
||||
|
||||
// InstanceState is get's answer: the instance's state and the sequence of the last operation it holds.
|
||||
type InstanceState struct {
|
||||
Instance string `json:"instance"`
|
||||
State json.RawMessage `json:"state"`
|
||||
Sequence uint64 `json:"sequence"`
|
||||
}
|
||||
|
||||
// HistoryRequest reads an instance's operations from a sequence on, at most Limit of them.
|
||||
type HistoryRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Instance string `json:"instance"`
|
||||
From uint64 `json:"from,omitempty"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
// Operation is one operation as the history keeps it.
|
||||
type Operation struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Instance string `json:"instance"`
|
||||
Operation string `json:"operation"`
|
||||
Body json.RawMessage `json:"body,omitempty"`
|
||||
Author Author `json:"author"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// HistoryPage is history's answer: the operations oldest first, the sequence to read on from (0 when nothing
|
||||
// follows) and the instance's last sequence when it was read.
|
||||
type HistoryPage struct {
|
||||
Operations []Operation `json:"operations"`
|
||||
Next uint64 `json:"next"`
|
||||
Last uint64 `json:"last"`
|
||||
}
|
||||
|
||||
// LedgerWatchRequest asks the tool runner to send every change of the instances Instances names (an instance, with
|
||||
// `*` for one name and `>` for the rest) as InstanceChange under the watch's id: the current states first.
|
||||
type LedgerWatchRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Instances string `json:"instances"`
|
||||
Watch string `json:"watch"`
|
||||
}
|
||||
|
||||
// InstanceChange is one instance's new state, sent under LedgerWatched; Current says it was the state when the watch
|
||||
// began.
|
||||
type InstanceChange struct {
|
||||
Watch string `json:"watch"`
|
||||
Instance string `json:"instance"`
|
||||
State json.RawMessage `json:"state"`
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Current bool `json:"current"`
|
||||
}
|
||||
|
||||
// QueryRequest reads a declared index of a ledger: the entries whose key Key names (a key, with `*` for one name and
|
||||
// `>` for the rest; empty is every key), at most Limit of them.
|
||||
type QueryRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Index string `json:"index"`
|
||||
Key string `json:"key,omitempty"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
// IndexEntry is the value of one key of a declared index (IndexKey): the instance it points at, and the sequence of
|
||||
// the operation it was written from. The owner's watcher and the tool runner's reindex both write it so.
|
||||
type IndexEntry struct {
|
||||
Instance string `json:"instance"`
|
||||
Sequence uint64 `json:"sequence"`
|
||||
}
|
||||
|
||||
// IndexKey is the key of an instance in a declared index: the values of the index's fields, in its order, then the
|
||||
// parts of the instance's path, all joined by ".". Each value and each path part is encoded (keyToken): every byte
|
||||
// outside [A-Za-z0-9_-] is written "=" and two upper-case hex digits ("." is "=2E", " " is "=20", "=" is "=3D"), so two
|
||||
// different keys never meet and every part is a token the bus takes. `17950.listing.a17` is the key of price 17950
|
||||
// and instance `listing/a17`. The owner's watcher and the tool runner's reindex both write it, so a reindex leaves the
|
||||
// index as the watcher keeps it.
|
||||
func IndexKey(fields []string, instance string) string {
|
||||
parts := make([]string, 0, len(fields)+2)
|
||||
for _, f := range fields {
|
||||
parts = append(parts, keyToken(f))
|
||||
}
|
||||
for _, p := range strings.Split(instance, "/") {
|
||||
parts = append(parts, keyToken(p))
|
||||
}
|
||||
return strings.Join(parts, ".")
|
||||
}
|
||||
|
||||
// keyToken is v with every byte outside [A-Za-z0-9_-] written "=" and two upper-case hex digits (IndexKey).
|
||||
func keyToken(v string) string {
|
||||
var b strings.Builder
|
||||
for i := 0; i < len(v); i++ {
|
||||
c := v[i]
|
||||
if c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-' {
|
||||
b.WriteByte(c)
|
||||
continue
|
||||
}
|
||||
fmt.Fprintf(&b, "=%02X", c)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// QueryAnswer is query's answer.
|
||||
type QueryAnswer struct {
|
||||
Entries []IndexEntry `json:"entries"`
|
||||
}
|
||||
|
||||
// RebuildFrom is what a rebuild derives the state from: the history, or the snapshot and the operations after it.
|
||||
const (
|
||||
RebuildFromHistory = "history"
|
||||
RebuildFromSnapshot = "snapshot"
|
||||
)
|
||||
|
||||
// RebuildRequest derives an instance's state again (every instance's when Instance is empty), From the history or the
|
||||
// snapshot.
|
||||
type RebuildRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Instance string `json:"instance,omitempty"`
|
||||
From string `json:"from,omitempty"`
|
||||
}
|
||||
|
||||
// ReindexRequest builds a declared index again from the history.
|
||||
type ReindexRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
Index string `json:"index"`
|
||||
}
|
||||
|
||||
// LedgerRequest names a ledger: lag.
|
||||
type LedgerRequest struct {
|
||||
Ledger string `json:"ledger"`
|
||||
}
|
||||
|
||||
// Repaired is the answer of rebuild, reindex and snapshot: how many instances (or index entries) it wrote.
|
||||
type Repaired struct {
|
||||
Written int `json:"written"`
|
||||
}
|
||||
|
||||
// InstanceLag is one instance whose state or snapshot is behind its history: the sequences each holds.
|
||||
type InstanceLag struct {
|
||||
Instance string `json:"instance"`
|
||||
History uint64 `json:"history"`
|
||||
State uint64 `json:"state"`
|
||||
Snapshot uint64 `json:"snapshot"`
|
||||
}
|
||||
|
||||
// IndexLag is how far one declared index is behind the history.
|
||||
type IndexLag struct {
|
||||
Index string `json:"index"`
|
||||
Behind uint64 `json:"behind"`
|
||||
}
|
||||
|
||||
// Lagged is lag's answer: every instance and index behind its history; none when nothing is behind.
|
||||
type Lagged struct {
|
||||
Instances []InstanceLag `json:"instances,omitempty"`
|
||||
Indexes []IndexLag `json:"indexes,omitempty"`
|
||||
}
|
||||
@@ -31,7 +31,8 @@ type ClaimedSeat struct {
|
||||
// WantedAccount is one key of the grants part's state `accounts`: an account the mesh needs, without its password.
|
||||
type WantedAccount struct {
|
||||
User string `json:"user"`
|
||||
// Kind is KindModule (KindNodeTools for the runtime, named as a module is), KindProcess or KindProvisioner.
|
||||
// Kind is KindModule (KindNodeTools for the runtime, named as a module is), KindProcess, KindProvisioner or
|
||||
// KindLedger.
|
||||
Kind Kind `json:"kind"`
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
@@ -46,8 +47,12 @@ type WantedAccount struct {
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// LedgersSecret is the tool runner's own secret holding its machine's ledger accounts, a map of module to credential
|
||||
// (ADR 0597 §9).
|
||||
const LedgersSecret = "ledgers"
|
||||
|
||||
// mintedKinds are the kinds of account the vault mints (ADR 0561 §1).
|
||||
var mintedKinds = []Kind{KindModule, KindNodeTools, KindProcess, KindProvisioner}
|
||||
var mintedKinds = []Kind{KindModule, KindNodeTools, KindProcess, KindProvisioner, KindLedger}
|
||||
|
||||
// principal is the account as the composer names it.
|
||||
func (w WantedAccount) principal() Principal {
|
||||
@@ -93,9 +98,12 @@ func (w WantedAccount) Same(o WantedAccount) bool {
|
||||
}
|
||||
|
||||
// OwnSecret is the module's own secret the account is held in: `broker`, `broker-<process id>` or `provisioner-broker`
|
||||
// (the controller's names, ADR 0436 §2, ADR 0349 §11).
|
||||
// (the controller's names, ADR 0436 §2, ADR 0349 §11); a ledger account in the tool runner's `ledgers`, which keeps every
|
||||
// ledger account of its machine, by module (ADR 0597 §9).
|
||||
func (w WantedAccount) OwnSecret() string {
|
||||
switch w.Kind {
|
||||
case KindLedger:
|
||||
return LedgersSecret
|
||||
case KindProcess:
|
||||
return "broker-" + w.Process
|
||||
case KindProvisioner:
|
||||
|
||||
@@ -18,4 +18,18 @@ func TestTheBusNamesAgree(t *testing.T) {
|
||||
if LogStreamName("mesh-delivery-builder", "build-log") == "" || LogSubject("mesh-delivery-builder", "build-log") == "" {
|
||||
t.Fatal("a log has no stream or subject")
|
||||
}
|
||||
// 1966/AC20: a ledger's history and snapshot subjects are one contract, which the grants part grants a ledger
|
||||
// account its publish on.
|
||||
if h, sn := LedgerHistorySubject("m", "records"), LedgerSnapshotSubject("m", "records"); h != "mesh.ledger.m.records.history" ||
|
||||
sn != "mesh.ledger.m.records.snapshot" {
|
||||
t.Fatalf("a ledger's subjects are %q and %q", h, sn)
|
||||
}
|
||||
// 1966/AC23: a wanted account of kind ledger for m on a names the user ledger.a.m and the own secret ledgers.
|
||||
ledger := WantedAccount{User: "ledger.a.m", Kind: KindLedger, Node: "a", Module: "m", URL: "nats://10.0.0.1:4222"}
|
||||
if got := (Principal{Kind: KindLedger, Node: "a", Module: "m"}).Username(); got != "ledger.a.m" {
|
||||
t.Fatalf("a ledger account is named %q", got)
|
||||
}
|
||||
if err := ledger.Valid(); err != nil || ledger.OwnSecret() != "ledgers" {
|
||||
t.Fatalf("a wanted ledger account: %v, held in %q", err, ledger.OwnSecret())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -44,8 +44,26 @@ const (
|
||||
// exactly what its bus says — never root, never the operator account, and never able to answer a tool
|
||||
// call or a seat's verb (§3). One per process and machine.
|
||||
KindProcess Kind = "process"
|
||||
// KindLedger is one module assignment's ledger credential (novox/hq ADR 0597 §9, design 25 § 12.4): held by that
|
||||
// machine's tool runner, never by the module's own code, and the only account that publishes on that module's ledger
|
||||
// histories and snapshots. One per module assignment, whose module asks for a ledger in `requires`.
|
||||
KindLedger Kind = "ledger"
|
||||
)
|
||||
|
||||
// LedgerHistorySubject and LedgerSnapshotSubject are the subject prefixes of one ledger's history and of its instances'
|
||||
// snapshots (novox/hq ADR 0597 §1, §8, §9): an instance's subject is the prefix, ".", and the instance's path with each
|
||||
// "/" written "." (`decision/0290` is `….decision.0290`). One contract: the grants part grants a ledger account its
|
||||
// publish on them, the tool runner appends and snapshots on them, and the bus module creates the history's stream over
|
||||
// them.
|
||||
func LedgerHistorySubject(module, local string) string {
|
||||
return "mesh.ledger." + module + "." + local + ".history"
|
||||
}
|
||||
|
||||
// LedgerSnapshotSubject is the subject prefix of one ledger's instances' snapshots (LedgerHistorySubject).
|
||||
func LedgerSnapshotSubject(module, local string) string {
|
||||
return "mesh.ledger." + module + "." + local + ".snapshot"
|
||||
}
|
||||
|
||||
// Seat is a role on the bus as a principal relates to it: the subjects it accepts, and those it
|
||||
// emits (novox/hq ADR 0118, design 29 §5).
|
||||
type Seat struct {
|
||||
@@ -272,6 +290,10 @@ func (p Principal) Username() string {
|
||||
return p.Login
|
||||
}
|
||||
return "node." + p.Node
|
||||
case KindLedger:
|
||||
// `ledger.<machine>.<module>`: never `<node>.<module>`, which is the module's own account or the tool runner's,
|
||||
// so the module's code never holds it.
|
||||
return "ledger." + p.Node + "." + p.Module
|
||||
case KindProvisioner:
|
||||
// Named for the machine, as a node is: one client per machine, whichever module runs it, and never
|
||||
// `<node>.<module>`, which is the module's own account or the tool runner's.
|
||||
|
||||
@@ -0,0 +1,207 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Bus state is what a module keeps on the bus, asked for as a provision of the bus module in `requires` (novox/hq ADR
|
||||
// 0595 §1, design 32 § 4): an entry of `requires` in its typed long form, the local name, the storage type, the template
|
||||
// and the template's options. This is the contract only (ADR 0578 §7, ADR 0593 §1): the entry's shape, decoded
|
||||
// leniently. Every rule that judges an entry (a template known, an option within its bound, a fixed setting not given)
|
||||
// is the indexer's judge's.
|
||||
|
||||
// The storage types (design 25 § 12.1): the stream is the base; the key-value bucket, the object store and the ledger
|
||||
// are built on it, each with calls of its own (ADR 0595 §2).
|
||||
const (
|
||||
StorageStream = "stream"
|
||||
StorageKeyValue = "key-value"
|
||||
StorageObjectStore = "object-store"
|
||||
StorageLedger = "ledger"
|
||||
)
|
||||
|
||||
// The six templates, each a storage type with a predefined configuration (ADR 0595 §2, design 25 § 12.1): state on
|
||||
// key-value, events, work queue and log on stream, objects on object store, ledger on ledger. A bare stream names
|
||||
// none and gives its own Configuration.
|
||||
const (
|
||||
TemplateState = "state"
|
||||
TemplateEvents = "events"
|
||||
TemplateWorkQueue = "work-queue"
|
||||
TemplateLog = "log"
|
||||
TemplateObjects = "objects"
|
||||
TemplateLedger = "ledger"
|
||||
)
|
||||
|
||||
// The option names, a template's settings a module may change (design 25 § 12.1's table), written as the old sections
|
||||
// wrote them, so a module that moves carries its old settings as options unchanged (ADR 0595 §11). Which option a
|
||||
// template takes, and its bound, is the judge's.
|
||||
const (
|
||||
// OptionPastValues is how many past values a state's key keeps (the old `state`'s `history`).
|
||||
OptionPastValues = "history"
|
||||
// OptionTimeToLive is how long a state's value lives, in seconds.
|
||||
OptionTimeToLive = "ttl-seconds"
|
||||
// OptionLifetimePerKey is whether a state's key may be given a lifetime of its own.
|
||||
OptionLifetimePerKey = "key-ttl"
|
||||
// OptionPerMachine is whether a state is kept one bucket per machine.
|
||||
OptionPerMachine = "per-machine"
|
||||
// OptionAgeDays is how long a stream keeps a message, in days.
|
||||
OptionAgeDays = "max-age-days"
|
||||
// OptionCapMiB is the cap of a stream, a bucket or an object store, in MiB.
|
||||
OptionCapMiB = "max-mib"
|
||||
// OptionGrant names the trusted config value granting another module a part of it (ADR 0363 §4, ADR 0592 §9).
|
||||
OptionGrant = "grant"
|
||||
// OptionDeliveries is how many times a work queue's message is delivered before it is a dead letter.
|
||||
OptionDeliveries = "max-deliveries"
|
||||
// OptionDuplicates is how long, in seconds, a work queue detects a message sent twice.
|
||||
OptionDuplicates = "duplicates"
|
||||
)
|
||||
|
||||
// ConfigWhenFull is what a configuration of its own says a bare stream does when full: WhenFullRefuse or
|
||||
// WhenFullDropOldest. In a template it is fixed, never an option (ADR 0595 §3).
|
||||
const (
|
||||
ConfigWhenFull = "when-full"
|
||||
WhenFullRefuse = "refuse"
|
||||
WhenFullDropOldest = "drop-oldest"
|
||||
)
|
||||
|
||||
// The ledger effects, what an operation does to an instance's state (ADR 0597 §2): one of this fixed set.
|
||||
const (
|
||||
// EffectSetFields sets the state's fields to the operation's Fields.
|
||||
EffectSetFields = "set-fields"
|
||||
// EffectSetEntry sets one entry of the state's map Into, under the key the operation's field Key holds, to its Fields.
|
||||
EffectSetEntry = "set-entry"
|
||||
// EffectAppend appends the operation's Fields to the state's list Into.
|
||||
EffectAppend = "append"
|
||||
// EffectRemove removes the entry of the map Into under the operation's field Key, or, naming no Into, the instance's
|
||||
// state.
|
||||
EffectRemove = "remove"
|
||||
)
|
||||
|
||||
// LedgerType is one type of a ledger (ADR 0597 §1): a schema and the operations allowed on it, with its indexes. An
|
||||
// instance of it is named by a path whose first part is the type's name (`decision/0290`).
|
||||
type LedgerType struct {
|
||||
Name string `json:"name"`
|
||||
// Schema is the type's fields, by name.
|
||||
Schema map[string]LedgerField `json:"schema"`
|
||||
// Operations are the operations allowed on an instance, each with its effect.
|
||||
Operations []LedgerOperation `json:"operations"`
|
||||
// Indexes are its declared indexes, each a second key-value bucket kept by a watcher (ADR 0597 §4).
|
||||
Indexes []LedgerIndex `json:"indexes,omitempty"`
|
||||
}
|
||||
|
||||
// LedgerField is one field of a ledger type's schema.
|
||||
type LedgerField struct {
|
||||
// Type is the field's value type, as the schema names it.
|
||||
Type string `json:"type"`
|
||||
// DigestOf names the text field this field is the digest of (ADR 0597 §2): `apply` refuses an operation whose
|
||||
// digest does not match its text.
|
||||
DigestOf string `json:"digest-of,omitempty"`
|
||||
}
|
||||
|
||||
// LedgerOperation is one operation a ledger type allows, and its effect on the state.
|
||||
type LedgerOperation struct {
|
||||
Name string `json:"name"`
|
||||
// Effect is one of the ledger effects (EffectSetFields, EffectSetEntry, EffectAppend, EffectRemove).
|
||||
Effect string `json:"effect"`
|
||||
// Fields are the operation's fields its effect copies into the state; it may carry more (ADR 0597 §2).
|
||||
Fields []string `json:"fields,omitempty"`
|
||||
// Into is the state's map (EffectSetEntry, EffectRemove) or list (EffectAppend) the effect writes.
|
||||
Into string `json:"into,omitempty"`
|
||||
// Key is the operation's field holding the entry's key (EffectSetEntry, EffectRemove).
|
||||
Key string `json:"key,omitempty"`
|
||||
// Sequence and Author are whether the effect also writes the operation's sequence and its proven author.
|
||||
Sequence bool `json:"sequence,omitempty"`
|
||||
Author bool `json:"author,omitempty"`
|
||||
}
|
||||
|
||||
// LedgerIndex is one declared index of a ledger type: its name, and the fields it is kept on.
|
||||
type LedgerIndex struct {
|
||||
Name string `json:"name"`
|
||||
On []string `json:"on"`
|
||||
}
|
||||
|
||||
// BusStateField is where a stored manifest keeps its bus-state entries: beside `requires`, which stays a plain list of
|
||||
// names, as OptionalRequiresKey keeps the optional ones.
|
||||
const BusStateField = "bus-state"
|
||||
|
||||
// BusStateEntry is one piece of a module's bus state, as its manifest asks for it.
|
||||
type BusStateEntry struct {
|
||||
// Name is its local name; the mesh, never the module, names it on the bus (ADR 0201 §4).
|
||||
Name string `json:"name"`
|
||||
// Storage is its storage type (design 25 § 12.1): how the bus keeps it.
|
||||
Storage string `json:"storage"`
|
||||
// Template is the predefined configuration of that storage type it uses; empty for a bare stream, which gives
|
||||
// Configuration instead.
|
||||
Template string `json:"template,omitempty"`
|
||||
// Options are the template's settings the module changes, by option name; each value as the manifest wrote it.
|
||||
Options map[string]json.RawMessage `json:"options,omitempty"`
|
||||
// Configuration is a configuration of the module's own, where no template fits: a bare stream's, or a ledger's
|
||||
// whose history expires (ADR 0595 §2).
|
||||
Configuration map[string]json.RawMessage `json:"configuration,omitempty"`
|
||||
// Types are a ledger's types (ADR 0597 §1), for the storage type ledger only.
|
||||
Types []LedgerType `json:"types,omitempty"`
|
||||
}
|
||||
|
||||
// isBusStateEntry is whether a `requires` item in its long form asks for bus state: it names a storage type.
|
||||
func isBusStateEntry(raw json.RawMessage) bool {
|
||||
var keys map[string]json.RawMessage
|
||||
if json.Unmarshal(raw, &keys) != nil {
|
||||
return false
|
||||
}
|
||||
_, has := keys["storage"]
|
||||
return has
|
||||
}
|
||||
|
||||
// readBusStateEntry reads one bus-state entry leniently (ADR 0578 §3): a key it does not know, at any depth, is set
|
||||
// aside and its path named, never refused. An error is only an entry of the wrong shape.
|
||||
func readBusStateEntry(raw json.RawMessage) (BusStateEntry, []string, error) {
|
||||
var tree any
|
||||
dec := json.NewDecoder(bytes.NewReader(raw))
|
||||
dec.UseNumber()
|
||||
if err := dec.Decode(&tree); err != nil {
|
||||
return BusStateEntry{}, nil, err
|
||||
}
|
||||
var aside []string
|
||||
strict := func(e *BusStateEntry) error {
|
||||
b, err := json.Marshal(tree)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d := json.NewDecoder(bytes.NewReader(b))
|
||||
d.DisallowUnknownFields()
|
||||
return typedUnknown(d.Decode(e))
|
||||
}
|
||||
for tries := 0; ; tries++ {
|
||||
var e BusStateEntry
|
||||
err := strict(&e)
|
||||
if err == nil {
|
||||
return e, aside, nil
|
||||
}
|
||||
unknown := asUnknownField(err)
|
||||
if unknown == nil || tries > 64 {
|
||||
return BusStateEntry{}, nil, err
|
||||
}
|
||||
path, ok := removalThatHelps(tree, unknown.Field, err.Error(), func() error { var e BusStateEntry; return strict(&e) })
|
||||
if !ok {
|
||||
if removeEvery(tree, unknown.Field) == 0 {
|
||||
return BusStateEntry{}, nil, err
|
||||
}
|
||||
path = "…." + unknown.Field
|
||||
}
|
||||
aside = append(aside, path)
|
||||
}
|
||||
}
|
||||
|
||||
// setAsideAlso adds paths read without to what a manifest set aside (SetAside), in its words.
|
||||
func setAsideAlso(unknown string, paths []string) string {
|
||||
if len(paths) == 0 {
|
||||
return unknown
|
||||
}
|
||||
if unknown == "" {
|
||||
field := paths[0][strings.LastIndex(paths[0], ".")+1:]
|
||||
unknown = fmt.Sprintf("json: unknown field %q", field)
|
||||
}
|
||||
return unknown + " (read without " + strings.Join(paths, ", ") + ")"
|
||||
}
|
||||
@@ -478,6 +478,10 @@ type Manifest struct {
|
||||
// RequireShares is who else may use the bucket a module requires, by requirement (BucketShare, novox/hq ADR 0468
|
||||
// §8): filled from the long form's `shares`, or from the stored form's `requires-shares`; never written by hand.
|
||||
RequireShares map[string][]BucketShare `json:"-"`
|
||||
// BusState is what the module keeps on the bus, each entry asked for in `requires` in its typed long form
|
||||
// (BusStateEntry, novox/hq ADR 0595 §1); its local name is never in Requires, which names provisions. Stored
|
||||
// beside `requires` (BusStateField), which stays a plain list of names.
|
||||
BusState []BusStateEntry `json:"bus-state,omitempty"`
|
||||
|
||||
// Emits are the events this module publishes, named **locally**: `order.placed`, not a subject
|
||||
// and not a routing key. The mesh derives where it lands (design 29 §1), so reorganising the
|
||||
@@ -1556,18 +1560,21 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
|
||||
}
|
||||
var optionalRequires map[string]bool
|
||||
var requireShares map[string][]BucketShare
|
||||
var busState []BusStateEntry
|
||||
var busStateAside []string
|
||||
if requires, ok := keys["requires"]; ok && string(requires) != "null" {
|
||||
names, optional, shares, err := readRequires(requires)
|
||||
read, err := readRequires(requires)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
plainNames, err := json.Marshal(names)
|
||||
plainNames, err := json.Marshal(read.names)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
keys["requires"] = plainNames
|
||||
optionalRequires = optional
|
||||
requireShares = shares
|
||||
optionalRequires = read.optional
|
||||
requireShares = read.shares
|
||||
busState, busStateAside = read.busState, read.aside
|
||||
}
|
||||
// The stored form's shares (RequireSharesKey).
|
||||
if stored, ok := keys[RequireSharesKey]; ok {
|
||||
@@ -1631,7 +1638,9 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
|
||||
}
|
||||
}
|
||||
*m = Manifest(fields)
|
||||
m.unknown = unknown
|
||||
m.unknown = setAsideAlso(unknown, busStateAside)
|
||||
// A bus-state entry of `requires`, beside those of the stored form (BusStateField).
|
||||
m.BusState = append(m.BusState, busState...)
|
||||
m.forms = forms
|
||||
m.OptionalRequires = optionalRequires
|
||||
m.RequireShares = requireShares
|
||||
|
||||
@@ -21,34 +21,59 @@ type requirementForm struct {
|
||||
Shares []BucketShare `json:"shares,omitempty"`
|
||||
}
|
||||
|
||||
// readRequires reads `requires` in both of its forms — a bare name, or {name, optional} — into the
|
||||
// names and which of them are optional.
|
||||
func readRequires(raw json.RawMessage) ([]string, map[string]bool, map[string][]BucketShare, error) {
|
||||
// requiresRead is `requires` as readRequires read it.
|
||||
type requiresRead struct {
|
||||
names []string
|
||||
optional map[string]bool
|
||||
shares map[string][]BucketShare
|
||||
// busState are the entries asking for bus state (BusStateEntry), never named in names, which name provisions; aside
|
||||
// the paths of the keys they had that this package does not know, set aside (ADR 0578 §3).
|
||||
busState []BusStateEntry
|
||||
aside []string
|
||||
}
|
||||
|
||||
// readRequires reads `requires` in its forms — a bare name, {name, optional}, or a bus-state entry (BusStateEntry,
|
||||
// novox/hq ADR 0595 §1) — into the names of the provisions, which of them are optional, and the bus-state entries,
|
||||
// whose local names stay off the names: a local name is no provision, and nothing provides it.
|
||||
func readRequires(raw json.RawMessage) (requiresRead, error) {
|
||||
var entries []json.RawMessage
|
||||
if err := json.Unmarshal(raw, &entries); err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("requires: a list of names, or of {name, optional}: %w", err)
|
||||
return requiresRead{}, fmt.Errorf("requires: a list of names, or of {name, optional}: %w", err)
|
||||
}
|
||||
var names []string
|
||||
optional := map[string]bool{}
|
||||
var shares map[string][]BucketShare
|
||||
var busState []BusStateEntry
|
||||
var aside []string
|
||||
for i, e := range entries {
|
||||
var plain string
|
||||
if err := json.Unmarshal(e, &plain); err == nil {
|
||||
names = append(names, plain)
|
||||
continue
|
||||
}
|
||||
if isBusStateEntry(e) {
|
||||
entry, unknown, err := readBusStateEntry(e)
|
||||
if err != nil {
|
||||
return requiresRead{}, fmt.Errorf("requires[%d]: a bus-state entry is {name, storage, template, options, configuration, types} (novox/hq ADR 0595 §1): %w", i, err)
|
||||
}
|
||||
busState = append(busState, entry)
|
||||
for _, path := range unknown {
|
||||
aside = append(aside, fmt.Sprintf("requires[%d]%s", i, path))
|
||||
}
|
||||
continue
|
||||
}
|
||||
var full requirementForm
|
||||
dec := json.NewDecoder(bytes.NewReader(e))
|
||||
dec.DisallowUnknownFields()
|
||||
if err := dec.Decode(&full); err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("requires[%d]: a name, or {name, optional} (novox/hq ADR 0330): %w", i, err)
|
||||
return requiresRead{}, fmt.Errorf("requires[%d]: a name, or {name, optional} (novox/hq ADR 0330): %w", i, err)
|
||||
}
|
||||
if strings.TrimSpace(full.Name) == "" {
|
||||
return nil, nil, nil, fmt.Errorf("requires[%d] names no requirement: {name, optional}", i)
|
||||
return requiresRead{}, fmt.Errorf("requires[%d] names no requirement: {name, optional}", i)
|
||||
}
|
||||
names = append(names, full.Name)
|
||||
if err := checkBucketShares(full.Name, full.Shares); err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("requires[%d]: %w", i, err)
|
||||
return requiresRead{}, fmt.Errorf("requires[%d]: %w", i, err)
|
||||
}
|
||||
if len(full.Shares) > 0 {
|
||||
if shares == nil {
|
||||
@@ -63,7 +88,7 @@ func readRequires(raw json.RawMessage) ([]string, map[string]bool, map[string][]
|
||||
if len(optional) == 0 {
|
||||
optional = nil
|
||||
}
|
||||
return names, optional, shares, nil
|
||||
return requiresRead{names: names, optional: optional, shares: shares, busState: busState, aside: aside}, nil
|
||||
}
|
||||
|
||||
// OptionalRequiresKey is where a stored manifest keeps which of its requirements are optional: beside
|
||||
|
||||
@@ -2,6 +2,7 @@ package catalogue
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
@@ -9,9 +10,8 @@ import (
|
||||
// A requirement a module can run without (novox/hq ADR 0330, issue 166). The shape of the issue:
|
||||
// grafana runs without a data source and requires `influxdb-api` softly; a mesh without influxdb
|
||||
// could not assign grafana at all, and one with influxdb in its catalogue would have had it installed
|
||||
// for grafana.
|
||||
|
||||
func TestARequirementIsHardUnlessMarkedOptional(t *testing.T) {
|
||||
// for grafana. And `requires` in its typed long form, a module's bus state (novox/hq ADR 0595 §1).
|
||||
func TestARequirementsLongForm(t *testing.T) {
|
||||
m := mustParse(t, `{"module":"grafana","version":"1",
|
||||
"requires":["route",{"name":"influxdb-api","optional":true},{"name":"oidc-client"}]}`)
|
||||
if strings.Join(m.Requires, " ") != "route influxdb-api oidc-client" {
|
||||
@@ -24,23 +24,12 @@ func TestARequirementIsHardUnlessMarkedOptional(t *testing.T) {
|
||||
t.Error("a requirement marked optional read as hard")
|
||||
}
|
||||
// And written back as it was read, so a manifest through the store keeps its form.
|
||||
raw, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var again Manifest
|
||||
if err := json.Unmarshal(raw, &again); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again := roundTrip(t, m)
|
||||
if !again.IsOptional("influxdb-api") || again.IsOptional("route") || len(again.Requires) != 3 {
|
||||
t.Fatalf("written and read back, it says %v optional %v (%s)", again.Requires, again.OptionalRequires, raw)
|
||||
t.Fatalf("written and read back, it says %v optional %v", again.Requires, again.OptionalRequires)
|
||||
}
|
||||
if strings.Contains(string(raw), `"name":`) {
|
||||
t.Errorf("a requirement was stored in the long form: %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestARequirementsLongFormRefusesWhatItDoesNotKnow(t *testing.T) {
|
||||
// The long form {name, optional} refuses what it does not know.
|
||||
for _, raw := range []string{
|
||||
`{"module":"x","version":"1","requires":[{"name":"influxdb-api","optinal":true}]}`,
|
||||
`{"module":"x","version":"1","requires":[{"optional":true}]}`,
|
||||
@@ -49,4 +38,94 @@ func TestARequirementsLongFormRefusesWhatItDoesNotKnow(t *testing.T) {
|
||||
t.Errorf("accepted %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
records := BusStateEntry{Name: "records", Storage: "ledger", Template: "ledger",
|
||||
Options: map[string]json.RawMessage{"max-mib": json.RawMessage(`1024`)}}
|
||||
// A ledger type with each of the four effects.
|
||||
listing := LedgerType{Name: "listing",
|
||||
Schema: map[string]LedgerField{"price": {Type: "number"}, "seen": {Type: "list"}, "notes": {Type: "map"},
|
||||
"text": {Type: "text"}, "digest": {Type: "text", DigestOf: "text"}},
|
||||
Operations: []LedgerOperation{
|
||||
{Name: "priced", Effect: EffectSetFields, Fields: []string{"price"}, Sequence: true, Author: true},
|
||||
{Name: "noted", Effect: EffectSetEntry, Fields: []string{"text", "digest"}, Into: "notes", Key: "title"},
|
||||
{Name: "seen", Effect: EffectAppend, Fields: []string{"at"}, Into: "seen"},
|
||||
{Name: "unnoted", Effect: EffectRemove, Into: "notes", Key: "title"},
|
||||
},
|
||||
Indexes: []LedgerIndex{{Name: "cheapest", On: []string{"price"}}}}
|
||||
typeJSON, err := json.Marshal(listing)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ledger := records
|
||||
ledger.Types = []LedgerType{listing}
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
requires string
|
||||
names string
|
||||
busState []BusStateEntry
|
||||
aside []string
|
||||
}{
|
||||
// 1966/AC1: a plain name and a bus-state entry: Requires names the plain one only, since it names provisions,
|
||||
// and BusState holds the entry whole.
|
||||
{"a plain name and a bus-state entry", `["postgres-database",
|
||||
{"name":"records","storage":"ledger","template":"ledger","options":{"max-mib":1024}}]`,
|
||||
"postgres-database", []BusStateEntry{records}, nil},
|
||||
// 1966/AC2: an unknown key is set aside and named, never refused; the entry decodes.
|
||||
{"a bus-state entry with a key it does not know", `[
|
||||
{"name":"records","storage":"ledger","template":"ledger","colour":"red","options":{"max-mib":1024}}]`,
|
||||
"", []BusStateEntry{records}, []string{"requires[0].colour"}},
|
||||
// 1966/AC3: a ledger type with each of the four effects round-trips through JSON unchanged.
|
||||
{"a ledger with its type", `[{"name":"records","storage":"ledger","template":"ledger","options":{"max-mib":1024},
|
||||
"types":[` + string(typeJSON) + `]}]`, "", []BusStateEntry{ledger}, nil},
|
||||
} {
|
||||
got, err := decodeForTest([]byte(`{"module":"x","version":"1","requires":` + c.requires + `}`))
|
||||
if err != nil {
|
||||
t.Errorf("%s: refused: %v", c.name, err)
|
||||
continue
|
||||
}
|
||||
if strings.Join(got.Requires, " ") != c.names {
|
||||
t.Errorf("%s: requires read as %v, want %s", c.name, got.Requires, c.names)
|
||||
}
|
||||
if !reflect.DeepEqual(got.BusState, c.busState) {
|
||||
t.Errorf("%s: bus state read as %+v, want %+v", c.name, got.BusState, c.busState)
|
||||
}
|
||||
for _, path := range c.aside {
|
||||
if !strings.Contains(got.SetAside(), path) {
|
||||
t.Errorf("%s: %s is not named among what was set aside: %q", c.name, path, got.SetAside())
|
||||
}
|
||||
}
|
||||
if len(c.aside) == 0 && got.SetAside() != "" {
|
||||
t.Errorf("%s: set aside %q", c.name, got.SetAside())
|
||||
}
|
||||
// Stored and read back, the entries stay whole and `requires` stays plain names.
|
||||
back := roundTrip(t, got)
|
||||
if !reflect.DeepEqual(back.BusState, c.busState) || strings.Join(back.Requires, " ") != c.names {
|
||||
t.Errorf("%s: read back as %v %+v", c.name, back.Requires, back.BusState)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// roundTrip is m written as the store keeps it and read again; the stored form keeps `requires` plain names.
|
||||
func roundTrip(t *testing.T, m Manifest) Manifest {
|
||||
t.Helper()
|
||||
raw, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var stored struct {
|
||||
Requires []any `json:"requires"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &stored); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, r := range stored.Requires {
|
||||
if _, plain := r.(string); !plain {
|
||||
t.Errorf("a requirement was stored in the long form: %s", raw)
|
||||
}
|
||||
}
|
||||
var again Manifest
|
||||
if err := json.Unmarshal(raw, &again); err != nil {
|
||||
t.Fatalf("%v: %s", err, raw)
|
||||
}
|
||||
return again
|
||||
}
|
||||
|
||||
@@ -75,7 +75,7 @@ func prunedFields(keys map[string]json.RawMessage) (manifestFields, []string, er
|
||||
removed = append(removed, k)
|
||||
break
|
||||
}
|
||||
path, ok := removalThatHelps(k, tree[k], unknown.Field, err.Error())
|
||||
path, ok := removalThatHelps(tree[k], unknown.Field, err.Error(), func() error { return decodesAlone(k, tree[k]) })
|
||||
if !ok {
|
||||
// The same key unknown in two entries alike: no one removal changes the words.
|
||||
// Every occurrence goes, and the field is judged again.
|
||||
@@ -113,8 +113,8 @@ func decodesAlone(k string, v any) error {
|
||||
}
|
||||
|
||||
// removalThatHelps removes, from v, the one occurrence of key whose removal changes what the strict
|
||||
// decoder says of field k, and says where it was. Every other occurrence is left as it was.
|
||||
func removalThatHelps(k string, v any, key, said string) (string, bool) {
|
||||
// decoder says (decodes, which decodes v strictly), and says where it was. Every other occurrence is left as it was.
|
||||
func removalThatHelps(v any, key, said string, decodes func() error) (string, bool) {
|
||||
var found bool
|
||||
var where string
|
||||
var walk func(node any, at string) bool
|
||||
@@ -126,7 +126,7 @@ func removalThatHelps(k string, v any, key, said string) (string, bool) {
|
||||
// Accepted only when the decoder is past it: nothing left, or an unknown key said
|
||||
// elsewhere. A different kind of error means the removal broke the entry; the same
|
||||
// words mean this was not the occurrence it refused.
|
||||
err := decodesAlone(k, v)
|
||||
err := decodes()
|
||||
if err == nil || (asUnknownField(err) != nil && err.Error() != said) {
|
||||
found, where = true, at+"."+key
|
||||
return true
|
||||
|
||||
@@ -5,6 +5,8 @@ import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// The verdicts on manifests (novox/hq ADR 0578 §1, §3): the indexer judges every manifest it takes in once, with its own
|
||||
@@ -28,6 +30,11 @@ const (
|
||||
// raises it as a condition beside the indexer's finding.
|
||||
ManifestRefusedEvent = "manifest-refused"
|
||||
|
||||
// ManifestAcceptedEvent is the indexer's event saying a manifest that asks for bus state was accepted
|
||||
// (ManifestAccepted), carrying its bus-state entries whole, so the bus module creates them without reading the
|
||||
// indexer's state (novox/hq ADR 0595 §4, design 25 § 12.2).
|
||||
ManifestAcceptedEvent = "manifest-accepted"
|
||||
|
||||
// ManifestOutgrown is the finding of a manifest the indexer accepted before and a stricter judge finds problems in
|
||||
// (novox/hq ADR 0579 §1): it stays accepted, its problems are the verdict's Warnings, and the watchman raises it as a
|
||||
// warning until the manifest is mended.
|
||||
@@ -154,3 +161,24 @@ type ManifestRefused struct {
|
||||
func RefusedOf(digest string, v ManifestVerdict) ManifestRefused {
|
||||
return ManifestRefused{V: Version, Module: v.Module, Digest: digest, Judge: v.Judge, Problems: v.Problems}
|
||||
}
|
||||
|
||||
// ManifestAccepted is the body of the indexer's event ManifestAcceptedEvent: the module, the digest of its manifest, the
|
||||
// judge, and every bus-state entry the manifest asks for, whole (novox/hq ADR 0595 §4).
|
||||
type ManifestAccepted struct {
|
||||
V int `json:"v"`
|
||||
Module string `json:"module"`
|
||||
Digest string `json:"digest"`
|
||||
Judge string `json:"judge"`
|
||||
BusState []catalogue.BusStateEntry `json:"bus-state"`
|
||||
}
|
||||
|
||||
// AcceptedOf is the event body saying v, the verdict accepting the manifest m of digest.
|
||||
func AcceptedOf(digest string, v ManifestVerdict, m catalogue.Manifest) ManifestAccepted {
|
||||
return ManifestAccepted{V: Version, Module: v.Module, Digest: digest, Judge: v.Judge, BusState: m.BusState}
|
||||
}
|
||||
|
||||
// EventID is the event's id, one per manifest digest and judge, so the bus takes it once:
|
||||
// `manifest-accepted-<digest>-<judge>`.
|
||||
func (a ManifestAccepted) EventID() string {
|
||||
return ManifestAcceptedEvent + "-" + a.Digest + "-" + a.Judge
|
||||
}
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// A verdict on a manifest round-trips, refuses what it cannot be, and a reader takes a missing one as accepted (novox/hq
|
||||
@@ -72,4 +76,28 @@ func TestAManifestVerdictIsTheContractReadersTake(t *testing.T) {
|
||||
if r, _ := TakeManifestVerdict(m.VerdictOf("x")); r || len(m.VerdictOf("x").Warnings) != 1 {
|
||||
t.Fatalf("an outgrown module: %+v", m.VerdictOf("x"))
|
||||
}
|
||||
|
||||
// 1966/AC6: the accepted event of a manifest with two bus-state entries carries both whole, and its id is
|
||||
// manifest-accepted-<digest>-<judge>.
|
||||
entries := []catalogue.BusStateEntry{
|
||||
{Name: "drafts", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState,
|
||||
Options: map[string]json.RawMessage{catalogue.OptionPastValues: json.RawMessage(`5`)}},
|
||||
{Name: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger,
|
||||
Types: []catalogue.LedgerType{{Name: "decision", Operations: []catalogue.LedgerOperation{{Name: "submit", Effect: catalogue.EffectSetFields}}}}},
|
||||
}
|
||||
digest := strings.Repeat("d", 64)
|
||||
accepted := AcceptedOf(digest, ManifestVerdict{Module: "mesh-knowledge", State: VerdictAccepted, Judge: "src2:j"},
|
||||
catalogue.Manifest{Module: "mesh-knowledge", BusState: entries})
|
||||
raw, err := json.Marshal(accepted)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var heard ManifestAccepted
|
||||
if err := json.Unmarshal(raw, &heard); err != nil || heard.Module != "mesh-knowledge" || heard.Digest != digest ||
|
||||
heard.V != Version || !reflect.DeepEqual(heard.BusState, entries) {
|
||||
t.Fatalf("the accepted event reads as %+v (%v)", heard, err)
|
||||
}
|
||||
if id := heard.EventID(); id != "manifest-accepted-"+digest+"-src2:j" {
|
||||
t.Fatalf("the accepted event's id is %q", id)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,8 @@ package membership
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// AssignmentsStream holds every assignment's membership, the newest per subject.
|
||||
@@ -57,6 +59,36 @@ type Membership struct {
|
||||
Hears []HeardIssued `json:"hears,omitempty"`
|
||||
// Waits are the event families a session may wait on through the mesh MCP server (novox/hq ADR 0367 §2).
|
||||
Waits []WaitIssued `json:"waits,omitempty"`
|
||||
// Bindings are the module's bus state as the bus module created it, one per entry its manifest asks for in
|
||||
// `requires` (novox/hq ADR 0595 §1, §7): written by the grants part when it hears the bus module's `created`.
|
||||
Bindings []BusBinding `json:"bindings,omitempty"`
|
||||
}
|
||||
|
||||
// BusBinding is one bus-state entry's names on the bus, by the local name its module asks for it by (novox/hq ADR 0595
|
||||
// §1, design 25 § 12.2): the module names only locally, and reaches its own entries, and only those, through these.
|
||||
// Which names are set follows the storage type: a stream's Stream and Subject, a key-value bucket's Bucket, an object
|
||||
// store's ObjectStore, and a ledger's History, State, Indexes and Snapshot.
|
||||
type BusBinding struct {
|
||||
Local string `json:"local"`
|
||||
// Storage and Template are the entry's storage type and template (catalogue.Storage…, catalogue.Template…).
|
||||
Storage string `json:"storage"`
|
||||
Template string `json:"template,omitempty"`
|
||||
// Stream and Subject are a stream's name and the subject its messages are on; Subject is also a ledger's history's.
|
||||
Stream string `json:"stream,omitempty"`
|
||||
Subject string `json:"subject,omitempty"`
|
||||
// Bucket is a key-value bucket's name.
|
||||
Bucket string `json:"bucket,omitempty"`
|
||||
// ObjectStore is an object store's name.
|
||||
ObjectStore string `json:"object-store,omitempty"`
|
||||
// Ledger, for the storage type ledger: its history stream (one subject per instance), its state bucket, each
|
||||
// declared index's bucket by index name, and the subject of the instances' snapshots (ADR 0597 §1, §4, §8).
|
||||
History string `json:"history,omitempty"`
|
||||
State string `json:"state,omitempty"`
|
||||
Indexes map[string]string `json:"indexes,omitempty"`
|
||||
Snapshot string `json:"snapshot,omitempty"`
|
||||
// Types are a ledger's types, copied by the grants part from the entry the manifest-accepted event carries, so the
|
||||
// tool runner checks an operation against its type without reading another part's state (ADR 0596, ADR 0597 §1).
|
||||
Types []catalogue.LedgerType `json:"types,omitempty"`
|
||||
}
|
||||
|
||||
// Served is one address a tool is answered on.
|
||||
|
||||
@@ -4,33 +4,26 @@ import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// A credential is read out of a bus address, and the bare address carries no credential and a scheme.
|
||||
func TestACredentialIsReadOutOfABusAddress(t *testing.T) {
|
||||
for _, c := range []struct{ in, user, password, bare string }{
|
||||
{"nats://controller:secret@127.0.0.1:4222", "controller", "secret", "nats://127.0.0.1:4222"},
|
||||
{"controller:secret@127.0.0.1:4222", "controller", "secret", "127.0.0.1:4222"},
|
||||
{"nats://127.0.0.1:4222", "", "", "nats://127.0.0.1:4222"},
|
||||
{"127.0.0.1:4222", "", "", "127.0.0.1:4222"},
|
||||
for _, c := range []struct{ in, user, password, bare, schemed string }{
|
||||
{"nats://controller:secret@127.0.0.1:4222", "controller", "secret", "nats://127.0.0.1:4222", "nats://127.0.0.1:4222"},
|
||||
{"controller:secret@127.0.0.1:4222", "controller", "secret", "127.0.0.1:4222", "nats://127.0.0.1:4222"},
|
||||
{"nats://127.0.0.1:4222", "", "", "nats://127.0.0.1:4222", "nats://127.0.0.1:4222"},
|
||||
{"127.0.0.1:4222", "", "", "127.0.0.1:4222", "nats://127.0.0.1:4222"},
|
||||
// A password containing an at-sign: split on the last one.
|
||||
{"nats://controller:a@b@127.0.0.1:4222", "controller", "a@b", "nats://127.0.0.1:4222"},
|
||||
{"nats://controller:a@b@127.0.0.1:4222", "controller", "a@b", "nats://127.0.0.1:4222", "nats://127.0.0.1:4222"},
|
||||
} {
|
||||
user, password, bare := CredentialIn(c.in)
|
||||
if user != c.user || password != c.password || bare != c.bare {
|
||||
t.Errorf("%q read as %q/%q at %q; wanted %q/%q at %q", c.in, user, password, bare, c.user, c.password, c.bare)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestABareAddressCarriesNoCredentialAndAScheme(t *testing.T) {
|
||||
for in, want := range map[string]string{
|
||||
"nats://controller:secret@10.0.0.1:4222": "nats://10.0.0.1:4222",
|
||||
"controller:secret@10.0.0.1:4222": "nats://10.0.0.1:4222",
|
||||
"nats://10.0.0.1:4222": "nats://10.0.0.1:4222",
|
||||
"10.0.0.1:4222": "nats://10.0.0.1:4222",
|
||||
} {
|
||||
if got := BareAddress(in); got != want {
|
||||
t.Errorf("BareAddress(%q) = %q, want %q", in, got, want)
|
||||
if got := BareAddress(c.in); got != c.schemed {
|
||||
t.Errorf("BareAddress(%q) = %q, want %q", c.in, got, c.schemed)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -66,6 +59,26 @@ func TestAMembershipRoundTripsAsIssued(t *testing.T) {
|
||||
if !reflect.DeepEqual(in, out) {
|
||||
t.Fatalf("round trip changed it:\n%+v\n%+v", in, out)
|
||||
}
|
||||
// 1966/AC4: a membership with a ledger binding, its types among it, round-trips; one without bindings encodes as before, byte for byte.
|
||||
bound := in
|
||||
bound.Bindings = []BusBinding{{Local: "records", Storage: "ledger", Template: "ledger", History: "forge_records",
|
||||
Subject: "forge_records.>", State: "forge_records_state", Indexes: map[string]string{"search": "forge_records_search"},
|
||||
Snapshot: "forge_records_snapshot.>",
|
||||
Types: []catalogue.LedgerType{{Name: "decision", Schema: map[string]catalogue.LedgerField{"title": {Type: "text"}},
|
||||
Operations: []catalogue.LedgerOperation{{Name: "propose", Effect: catalogue.EffectSetFields, Fields: []string{"title"}}}}}}}
|
||||
boundRaw, err := json.Marshal(bound)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var boundBack Membership
|
||||
if err := json.Unmarshal(boundRaw, &boundBack); err != nil || !reflect.DeepEqual(bound, boundBack) {
|
||||
t.Fatalf("a membership with a ledger binding read back as %+v (%v)", boundBack, err)
|
||||
}
|
||||
before, _ := json.Marshal(Membership{Node: "ace", Module: "forge", Tools: "t", Emits: "e", Serves: []Served{{Subject: "s"}},
|
||||
State: []StateIssued{{Name: "repos", Bucket: "forge_repos", Writes: true}}})
|
||||
if string(before) != `{"node":"ace","module":"forge","serves":[{"subject":"s"}],"emits":"e","tools":"t","state":[{"name":"repos","bucket":"forge_repos","writes":true}]}` {
|
||||
t.Fatalf("a membership without bindings encodes otherwise than before: %s", before)
|
||||
}
|
||||
var wire map[string]json.RawMessage
|
||||
_ = json.Unmarshal(raw, &wire)
|
||||
for _, k := range []string{"receives", "mesh", "seat-traffic"} {
|
||||
|
||||
@@ -1,14 +1,19 @@
|
||||
package stdio
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"go/ast"
|
||||
"go/parser"
|
||||
"go/token"
|
||||
"io/fs"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||
)
|
||||
|
||||
// meshSDKPackages is every package mesh-sdk holds (novox/hq ADR 0593 §1): what every module shares, the value shapes
|
||||
@@ -16,7 +21,7 @@ import (
|
||||
// Logic only some modules need is its owner module's package. A package added here is a decision to say in the pull
|
||||
// request: does every module share it, or is it a shape more than one part reads?
|
||||
var meshSDKPackages = []string{
|
||||
".", "announce", "announcetest", "asker", "asks", "buildstep", "buscompose", "busretry", "calladdress", "catalogue",
|
||||
".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busretry", "calladdress", "catalogue",
|
||||
"composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership",
|
||||
"objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions",
|
||||
"sdkversion", "seal", "testbus",
|
||||
@@ -56,6 +61,16 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// 1966/AC5: the calls package imports nothing of mesh-sdk but the catalogue's types, so a change here rebuilds
|
||||
// only its importers (ADR 0595 §9).
|
||||
if dir == "buscalls" {
|
||||
for _, imp := range file.Imports {
|
||||
p, _ := strconv.Unquote(imp.Path.Value)
|
||||
if strings.HasPrefix(p, sdkModule) && p != sdkModule+"/catalogue" {
|
||||
t.Errorf("%s imports %s: the calls package imports nothing of mesh-sdk but catalogue (ADR 0595 §9)", path, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, decl := range file.Decls {
|
||||
fn, ok := decl.(*ast.FuncDecl)
|
||||
if !ok {
|
||||
@@ -90,4 +105,60 @@ func TestMeshSDKHoldsOnlyWhatEveryModuleShares(t *testing.T) {
|
||||
t.Errorf("mesh-sdk's list keeps %s, which is gone: take it off the list", want)
|
||||
}
|
||||
}
|
||||
|
||||
// 1966/AC5: the call an apply names is the id the tool runner hands under `_meta`'s mesh/call.
|
||||
if buscalls.MetaCall != "mesh/call" {
|
||||
t.Errorf("the tool call's id is handed under %q, not mesh/call", buscalls.MetaCall)
|
||||
}
|
||||
// 1966/AC5: an index's key is its fields' values, then the instance's path, joined by "."; its entry names the
|
||||
// instance and the sequence.
|
||||
for _, c := range []struct {
|
||||
fields []string
|
||||
instance string
|
||||
want string
|
||||
}{
|
||||
{[]string{"17950", "volvo"}, "listing/a17", "17950.volvo.listing.a17"},
|
||||
// A value holding "." and one that does not meet in no key.
|
||||
{[]string{"a.b"}, "c", "a=2Eb.c"},
|
||||
{[]string{"a"}, "b/c", "a.b.c"},
|
||||
// A title with spaces is a token the bus takes; "=" itself is encoded.
|
||||
{[]string{"How it is checked", "x=y"}, "decision/0290", "How=20it=20is=20checked.x=3Dy.decision.0290"},
|
||||
} {
|
||||
k := buscalls.IndexKey(c.fields, c.instance)
|
||||
if k != c.want {
|
||||
t.Errorf("the index key of %q and %q is %q, want %q", c.fields, c.instance, k, c.want)
|
||||
}
|
||||
for _, token := range strings.Split(k, ".") {
|
||||
if token == "" || strings.ContainsAny(token, " *>\t") {
|
||||
t.Errorf("the index key %q holds a token the bus refuses: %q", k, token)
|
||||
}
|
||||
}
|
||||
}
|
||||
if buscalls.IndexKey([]string{"a.b"}, "c") == buscalls.IndexKey([]string{"a"}, "b/c") {
|
||||
t.Error("two different index keys meet")
|
||||
}
|
||||
entry, _ := json.Marshal(buscalls.IndexEntry{Instance: "listing/a17", Sequence: 3})
|
||||
if string(entry) != `{"instance":"listing/a17","sequence":3}` {
|
||||
t.Errorf("an index entry is %s", entry)
|
||||
}
|
||||
// 1966/AC5: an apply naming the tool call it serves, and its answer with the state behind, round-trip.
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
in, back any
|
||||
}{
|
||||
{"apply", &buscalls.ApplyRequest{Ledger: "records", Instance: "decision/0290", Operation: "submit",
|
||||
Body: json.RawMessage(`{"why":"x"}`), Call: "call-17"}, &buscalls.ApplyRequest{}},
|
||||
{"applied", &buscalls.Applied{Taken: true, Sequence: 5, StateBehind: true}, &buscalls.Applied{}},
|
||||
} {
|
||||
raw, err := json.Marshal(c.in)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := json.Unmarshal(raw, c.back); err != nil || !reflect.DeepEqual(c.in, c.back) {
|
||||
t.Errorf("%s read back as %+v from %s (%v)", c.name, c.back, raw, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sdkModule is mesh-sdk's Go module path.
|
||||
const sdkModule = "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
Reference in New Issue
Block a user