One bad module no longer drops its machine, too-large values and bus blackouts fail loud or are waited out, watches are never forgotten (issue 774) #138
@@ -42,16 +42,12 @@ func ControllerAnnouncement(node string) announce.States {
|
||||
return s
|
||||
}
|
||||
|
||||
// ReadsAnnounced splits a reader's `reads` (`<owner>.<name>`) by what each owner announced: granted when the owner
|
||||
// announced the state readable, or announced nothing yet (the transition); refused when the owner announced and did
|
||||
// not announce that state readable.
|
||||
func ReadsAnnounced(reads []string, announced map[string]announce.States) (granted, refused []string) {
|
||||
return ReadsAnnouncedBy("", reads, announced)
|
||||
}
|
||||
|
||||
// ReadsAnnouncedBy is ReadsAnnounced for one reader module, which also honours the readers an owner named for a state
|
||||
// (announce.State.Readers) and the composer's named readers (NamedReaders, ADR 0547 clause 9): a read named for other
|
||||
// modules is refused. An empty reader is judged by the readable mark alone.
|
||||
// ReadsAnnouncedBy splits one reader module's `reads` (`<owner>.<name>`) by what each owner announced: granted when the
|
||||
// owner announced the state readable (by this reader, where it named its readers: announce.State.Readers), when the
|
||||
// composer names the reader among the state's readers (NamedReaders, ADR 0547 clause 9), or when the owner announced
|
||||
// nothing yet (the transition); refused otherwise, and a read named for other modules is refused. There is no judgement
|
||||
// without the reader (issue 774): one made without it refused every named reader, and Users judges again for the
|
||||
// reader whatever split it is given (withAnnouncedReads). An empty reader is judged by the readable mark alone.
|
||||
func ReadsAnnouncedBy(reader string, reads []string, announced map[string]announce.States) (granted, refused []string) {
|
||||
for _, r := range reads {
|
||||
owner, name, ok := strings.Cut(r, ".")
|
||||
|
||||
@@ -552,6 +552,34 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
|
||||
t.Errorf("%s reads the lease: %v, want %v", reader, granted, want)
|
||||
}
|
||||
}
|
||||
// Issue 774 (sweep item 2): Users judges every read for its reader again, whatever split the caller made: the
|
||||
// controller's own reader-less copy refused the vault the lease and the grants part's accounts it is named for.
|
||||
announced["mesh-grants"] = announce.States{Module: "mesh-grants", States: []announce.State{{Name: "accounts", Readers: []string{VaultSeat}}}}
|
||||
announced["disk-load"] = announce.States{Module: "disk-load", States: []announce.State{{Name: "readings"}}}
|
||||
judged, err := Users(Records{Nodes: []string{"a"}, Announced: announced, Assigned: map[string][]Declared{"a": {
|
||||
{Module: VaultSeat, ReadsRefused: []string{ControllerLease, GrantsAccounts}},
|
||||
{Module: "mesh-config", ReadsRefused: []string{ControllerLease}},
|
||||
{Module: "restic", ReadsRefused: []string{"disk-load.readings"}},
|
||||
{Module: "messenger", Reads: []string{ControllerLease, GrantsAccounts}}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reads := map[string]string{}
|
||||
for _, u := range judged {
|
||||
if u.Kind == KindModule {
|
||||
reads[u.Module] = strings.Join(u.Reads, ",")
|
||||
}
|
||||
}
|
||||
if reads[VaultSeat] != ControllerLease+","+GrantsAccounts || reads["mesh-config"] != ControllerLease || reads["restic"] != "" || reads["messenger"] != "" {
|
||||
t.Errorf("the reads granted for their readers: %v", reads)
|
||||
}
|
||||
// AccountsEntry is the very lines ComposeAccounts writes for a user, so a writer of kept users repeats none.
|
||||
filled, _ := WithPasswords(judged, map[string]string{judged[0].Username(): "$2a$11$h"})
|
||||
list, err := ComposeAccounts(filled[:1])
|
||||
perms, perr := PermissionsFor(filled[0])
|
||||
if err != nil || perr != nil || !strings.Contains(list, " users = [\n"+AccountsEntry(filled[0].Username(), "$2a$11$h", perms)+" ]\n") {
|
||||
t.Errorf("the accounts entry is not ComposeAccounts' own: %v %v\n%s", err, perr, list)
|
||||
}
|
||||
c, _ := ConsumerFor(Principal{Kind: KindModule, Node: "a", Module: "m", Consumes: []string{"x.happened"}})
|
||||
cfg := ConsumerConfig(c)
|
||||
if cfg.Durable != "a_m" || cfg.AckPolicy != jetstream.AckExplicitPolicy || cfg.AckWait != 30*time.Second ||
|
||||
|
||||
+20
-11
@@ -1680,17 +1680,7 @@ func ComposeAccounts(principals []Principal) (string, error) {
|
||||
if p.PasswordHash == "" {
|
||||
return "", fmt.Errorf("%s has no password hash: a user without one is a user anybody is", p.Username())
|
||||
}
|
||||
fmt.Fprintf(&b, " { user: %q, password: %q, permissions: {\n", p.Username(), p.PasswordHash)
|
||||
if len(perms.PublishDeny) > 0 {
|
||||
fmt.Fprintf(&b, " publish: { allow: [%s], deny: [%s] }\n", quoted(perms.Publish), quoted(perms.PublishDeny))
|
||||
} else {
|
||||
fmt.Fprintf(&b, " publish: { allow: [%s] }\n", quoted(perms.Publish))
|
||||
}
|
||||
fmt.Fprintf(&b, " subscribe: { allow: [%s] }\n", quoted(perms.Subscribe))
|
||||
if perms.AllowResponses {
|
||||
fmt.Fprintf(&b, " allow_responses: { max: 1, ttl: \"%dm\" }\n", int(ResponseTTL/time.Minute))
|
||||
}
|
||||
b.WriteString(" } }\n")
|
||||
b.WriteString(AccountsEntry(p.Username(), p.PasswordHash, perms))
|
||||
}
|
||||
b.WriteString(" ]\n }\n")
|
||||
b.WriteString(systemAccount)
|
||||
@@ -1698,6 +1688,25 @@ func ComposeAccounts(principals []Principal) (string, error) {
|
||||
return b.String(), nil
|
||||
}
|
||||
|
||||
// AccountsEntry is one user of the accounts block, as ComposeAccounts writes each: its name, its password hash and its
|
||||
// permissions, in the lines the bus reads. Exported so a writer that keeps a user from grants it holds rather than
|
||||
// composes (mesh-grants' refused users) writes the same lines, never a copy of them (issue 774).
|
||||
func AccountsEntry(user, passwordHash string, perms Permissions) string {
|
||||
var b strings.Builder
|
||||
fmt.Fprintf(&b, " { user: %q, password: %q, permissions: {\n", user, passwordHash)
|
||||
if len(perms.PublishDeny) > 0 {
|
||||
fmt.Fprintf(&b, " publish: { allow: [%s], deny: [%s] }\n", quoted(perms.Publish), quoted(perms.PublishDeny))
|
||||
} else {
|
||||
fmt.Fprintf(&b, " publish: { allow: [%s] }\n", quoted(perms.Publish))
|
||||
}
|
||||
fmt.Fprintf(&b, " subscribe: { allow: [%s] }\n", quoted(perms.Subscribe))
|
||||
if perms.AllowResponses {
|
||||
fmt.Fprintf(&b, " allow_responses: { max: 1, ttl: \"%dm\" }\n", int(ResponseTTL/time.Minute))
|
||||
}
|
||||
b.WriteString(" } }\n")
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// WatchmanModule is the one module whose principal hears the bus's connect and disconnect events and asks its list of
|
||||
// connections (novox/hq ADR 0431 §2, issue 1147): the watchman of the repository mesh-delivery, which writes presence
|
||||
// from them. No other principal is granted the subjects below, and nothing is a wildcard over them.
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/handacts"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
@@ -83,6 +84,9 @@ const VaultBusAccounts = "mesh-vault.bus-accounts"
|
||||
// ArtifactStoreTools is the module holding the seat mesh-artifact-sweep (novox/hq ADR 0571 clause 2).
|
||||
const ArtifactStoreTools = "artifact-store-tools"
|
||||
|
||||
// ConfigModule is the module that holds the seat mesh-config, a reader of the controller's lease (NamedReaders).
|
||||
const ConfigModule = "mesh-config"
|
||||
|
||||
// NamedReaders are the states only the modules named may read, by `<owner>.<name>`: the grants part's wanted accounts
|
||||
// are read by the vault alone (novox/hq ADR 0561 §2).
|
||||
// The controller's lease is read by the parts that turn on at its write (ADR 0561 §3, §6), and by the assigner, which
|
||||
@@ -94,7 +98,9 @@ const ArtifactStoreTools = "artifact-store-tools"
|
||||
var NamedReaders = map[string][]string{VaultBusAccounts: {GrantsModule}, GrantsAccounts: {VaultSeat},
|
||||
ControllerLease: {VaultSeat, GrantsModule, delivery.AssignerModule, health.KeeperModule, health.HealersModule, WatchmanModule,
|
||||
// ADR 0571 clause 2: the artifact store's seat sweeps only once the lease names no controller.
|
||||
ArtifactStoreTools},
|
||||
ArtifactStoreTools,
|
||||
// ADR 0561 §3: mesh-config takes over the config only once the lease is handed away (its takeover.go leaseAway).
|
||||
ConfigModule},
|
||||
ControllerConditions: {health.KeeperModule}, ControllerHandActs: {health.HealersModule, handacts.Module}, HealersWould: {}}
|
||||
|
||||
// ReaderMay says whether module may read the state read (`<owner>.<name>`) by the named readers: always, for a state
|
||||
@@ -116,6 +122,23 @@ func readsNamed(module string, reads []string) (granted, refused []string) {
|
||||
return granted, refused
|
||||
}
|
||||
|
||||
// withAnnouncedReads is d with its reads judged for d's module (ReadsAnnouncedBy) against what the owners announced:
|
||||
// every read its manifest names, granted or refused by the caller, judged again here, so a caller that split them
|
||||
// without the reader (the controller's own copy of a reader-less judgement, issue 774: the vault, the assigner, the
|
||||
// grants part and the artifact store's tools refused the lease they are named for) cannot refuse a read the reader is
|
||||
// named for, nor grant one named for another. With no announcements read it is withNamedReaders.
|
||||
func withAnnouncedReads(d Declared, announced map[string]announce.States) Declared {
|
||||
if announced == nil {
|
||||
return withNamedReaders(d)
|
||||
}
|
||||
all := append(append([]string(nil), d.Reads...), d.ReadsRefused...)
|
||||
if len(all) == 0 {
|
||||
return d
|
||||
}
|
||||
d.Reads, d.ReadsRefused = ReadsAnnouncedBy(d.Module, all, announced)
|
||||
return d
|
||||
}
|
||||
|
||||
// withNamedReaders is d with the reads named for other modules moved to its refused reads.
|
||||
func withNamedReaders(d Declared) Declared {
|
||||
granted, refused := readsNamed(d.Module, d.Reads)
|
||||
|
||||
@@ -205,9 +205,9 @@ func Users(r Records) ([]Principal, error) {
|
||||
if !Composed(d) {
|
||||
continue
|
||||
}
|
||||
// A state with named readers is read by them alone (ADR 0547 clause 9): a read named for another module is
|
||||
// granted nothing, here and in the runtime carrying the module.
|
||||
d = withReadLogs(withNamedReaders(d), logs)
|
||||
// A state with named readers is read by them alone (ADR 0547 clause 9), and a read is granted what its owner
|
||||
// announced (ADR 0495 clause 3): judged here, for this reader, whatever split the caller made (issue 774).
|
||||
d = withReadLogs(withAnnouncedReads(d, r.Announced), logs)
|
||||
// **A process account is its own, delivered in its own secret** (broker-<process id>, novox/hq
|
||||
// ADR 0436 §1, §2), never the module's `broker` — so it is composed whether or not the module
|
||||
// itself has an account, and is never carried by the machine's runtime, which runs as the
|
||||
@@ -259,7 +259,7 @@ func Users(r Records) ([]Principal, error) {
|
||||
var carried []Declared
|
||||
for _, d := range r.Assigned[node] {
|
||||
if CarriedByRunner(d) {
|
||||
d = withReadLogs(withNamedReaders(d), logs)
|
||||
d = withReadLogs(withAnnouncedReads(d, r.Announced), logs)
|
||||
// Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1).
|
||||
d.Hears = ComposedHears(d)
|
||||
carried = append(carried, d)
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
// Package busretry waits out a bus that answers nothing for a few seconds (issue 774, the sweep of 2026-10-30: bus
|
||||
// blackouts at every user-list reload). nats-server 2.11's reloadAuthorization rebuilds the accounts, and for a few
|
||||
// seconds the mesh's account has none of its imports, $JS.API.> among them: every request is answered "no responders"
|
||||
// and every JetStream call "jetstream not enabled". A write or a call made in that window failed, and was lost.
|
||||
//
|
||||
// Only an error that says the request reached no one is waited out (Undelivered): nothing was done, so asking again
|
||||
// is safe even for a write that is not idempotent. A timeout, or any answer, is never asked again: the request may
|
||||
// have been delivered. The wait is bounded (Bound, about 30 s, backing off), and said on standard error when it ends
|
||||
// without an answer.
|
||||
//
|
||||
// This is the quick fix. The thorough one is a reload that keeps the account's imports, in the bus's holder.
|
||||
package busretry
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// Bound is how long a request that reached no one is asked again; FirstWait the first pause, doubled to MostWait.
|
||||
var (
|
||||
Bound = 30 * time.Second
|
||||
FirstWait = 250 * time.Millisecond
|
||||
MostWait = 4 * time.Second
|
||||
)
|
||||
|
||||
// Undelivered says err means the request reached no one: no responders, JetStream not enabled (for the account), or
|
||||
// no response from the stream after the client's own retries on no responders; also said in words by a runtime that
|
||||
// passed the error on.
|
||||
func Undelivered(err error) bool {
|
||||
var answered *Answered
|
||||
if err == nil || errors.As(err, &answered) {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, jetstream.ErrJetStreamNotEnabled) ||
|
||||
errors.Is(err, jetstream.ErrJetStreamNotEnabledForAccount) || errors.Is(err, jetstream.ErrNoStreamResponse) ||
|
||||
errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) {
|
||||
return true
|
||||
}
|
||||
words := strings.ToLower(err.Error())
|
||||
return strings.Contains(words, "no responders") || strings.Contains(words, "jetstream not enabled")
|
||||
}
|
||||
|
||||
// Answered is an error a callee answered with: the request was delivered, whatever its words say ("jetstream not
|
||||
// enabled" in a callee's own answer), so Undelivered never counts it and Do never asks it again.
|
||||
type Answered struct{ Words string }
|
||||
|
||||
func (a *Answered) Error() string { return a.Words }
|
||||
|
||||
// Do runs try, and again while its error says the request reached no one, until Bound has passed. blackout, when
|
||||
// given, is asked after each such error and Do stops waiting when it says the bus is not blacked out: a call nobody
|
||||
// answers because its callee is not running fails at once, as before. what names the request in the error.
|
||||
func Do[T any](what string, try func() (T, error), blackout func() bool) (T, error) {
|
||||
v, err := try()
|
||||
if !Undelivered(err) || (blackout != nil && !blackout()) {
|
||||
return v, err
|
||||
}
|
||||
start, wait := time.Now(), FirstWait
|
||||
for time.Since(start) < Bound {
|
||||
time.Sleep(wait)
|
||||
wait = min(wait*2, MostWait)
|
||||
v, err = try()
|
||||
if !Undelivered(err) || (blackout != nil && !blackout()) {
|
||||
return v, err
|
||||
}
|
||||
}
|
||||
err = fmt.Errorf("%s: the bus answered nothing for %s, asked again until then (a user-list reload drops the "+
|
||||
"account's imports for a few seconds; this was longer): %w", what, time.Since(start).Round(time.Second), err)
|
||||
fmt.Fprintf(os.Stderr, "[mesh-sdk] %v\n", err)
|
||||
return v, err
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package busretry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// Issue 774 (bus blackouts at every user-list reload): only a request that reached no one is asked again, within the
|
||||
// bound; a blackout probe that says the bus answers ends the wait at once.
|
||||
func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) {
|
||||
Bound, FirstWait, MostWait = 60*time.Millisecond, time.Millisecond, 4*time.Millisecond
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
errs []error // what each try answers; the last repeats
|
||||
blackout func() bool
|
||||
tries int // at least
|
||||
atMost int
|
||||
delivered bool
|
||||
}{
|
||||
{"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true},
|
||||
{"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true},
|
||||
{"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true},
|
||||
{"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false},
|
||||
{"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false},
|
||||
{"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false},
|
||||
{"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 3, 1000, false},
|
||||
} {
|
||||
tries := 0
|
||||
_, err := Do("the test", func() (int, error) {
|
||||
e := row.errs[min(tries, len(row.errs)-1)]
|
||||
tries++
|
||||
return tries, e
|
||||
}, row.blackout)
|
||||
if tries < row.tries || tries > row.atMost || (err == nil) != row.delivered {
|
||||
t.Errorf("%s: %d tries, %v", row.name, tries, err)
|
||||
}
|
||||
if strings.HasPrefix(row.name, "blacked out") && (!Undelivered(err) || !strings.Contains(fmt.Sprint(err), "the bus answered nothing")) {
|
||||
t.Errorf("%s: %v", row.name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -27,14 +28,28 @@ func TestTheBussAccountSaysNothingOnTheBus(t *testing.T) {
|
||||
"claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"},
|
||||
"emits":["bus.slow-consumer"],"tools":["bus_streams"],
|
||||
"build":{"artifacts":[{"name":"tools","kind":"bundle","language":"go","system":"arch","from":"cmd/t","binary":"t"}]}}`)
|
||||
if _, err := ParseManifest(bundled); err != nil {
|
||||
m, err := ParseManifest(bundled)
|
||||
if err != nil {
|
||||
t.Fatalf("a bus module whose tools bundle declares what it emits was refused: %v", err)
|
||||
}
|
||||
// And so does its built manifest, its bundle in Bundles and its build read away (issue 774: nats's built manifest
|
||||
// was refused, and with it every plan of the control-node).
|
||||
built, err := m.Resolve([]Built{{Name: "tools", Kind: ArtifactBundle, Reference: "store/tools:1", Digest: "sha256:1"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, err = json.Marshal(built)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ParseManifestForInputs(raw); err != nil {
|
||||
t.Fatalf("the built manifest of a bus module whose tools bundle declares what it emits was refused: %v", err)
|
||||
}
|
||||
|
||||
quiet := []byte(`{"module":"bus","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],
|
||||
"claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"},
|
||||
"tools":["bus_streams"]}`)
|
||||
m, err := ParseManifest(quiet)
|
||||
m, err = ParseManifest(quiet)
|
||||
if err != nil {
|
||||
t.Fatalf("a bus module with an account and tools was refused: %v", err)
|
||||
}
|
||||
|
||||
@@ -3378,6 +3378,11 @@ func whileStoppedProblems(m Manifest, r map[string]any, scheduled bool) []string
|
||||
|
||||
// shipsABundle says the module builds a tools bundle, which the node's runtime carries and speaks for.
|
||||
func (m Manifest) shipsABundle() bool {
|
||||
// A built manifest says its bundles in Bundles, its build read away (Resolve; issue 774: nats's built manifest was
|
||||
// refused for the emits its tools bundle says).
|
||||
if len(m.Bundles) > 0 {
|
||||
return true
|
||||
}
|
||||
if m.Build == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -155,7 +155,9 @@ var stateKeyName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`)
|
||||
|
||||
// ReadsGranted is the state a holder reads because modules on its machine offer it in pieces of a
|
||||
// kind its seat receives as data (novox/hq ADR 0260), each with the one key the piece shows — the
|
||||
// machine's own when it names none — sorted and once each. The bus grants the holder these keys and
|
||||
// machine's own when it names none — sorted and once each. A module's items of such a kind (design 57 L3) grant
|
||||
// what they show too, always at the machine's own key; an item whose capability the machine lacks grants a key
|
||||
// that is never written, which is no wider than its own machine's. The bus grants the holder these keys and
|
||||
// no others: the holder names no contributor, and reads only what is offered to it where it runs.
|
||||
func ReadsGranted(holder Manifest, onMachine []Manifest, machine string) []StateRead {
|
||||
seen := map[StateRead]bool{}
|
||||
@@ -183,6 +185,21 @@ func ReadsGranted(holder Manifest, onMachine []Manifest, machine string) []State
|
||||
}
|
||||
}
|
||||
}
|
||||
// A moved contributor's items of the kind (design 57 L3): what each shows, read from its attributes as a
|
||||
// contribution's data, and always this machine's own key (ADR 0260), never another's (issue 774:
|
||||
// power's draw block became an item and i3status-rust was issued nothing).
|
||||
for _, it := range m.Items {
|
||||
if it.Collection != "contribution:"+s.Name+":"+r.Kind {
|
||||
continue
|
||||
}
|
||||
for _, sr := range r.Shape.Reads(itemData(it)) {
|
||||
sr.Key = machine
|
||||
if module, _, err := ReadState(sr.Read); err == nil && module == m.Module && module != holder.Module &&
|
||||
stateKeyName.MatchString(sr.Key) {
|
||||
seen[sr] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -199,6 +216,17 @@ func ReadsGranted(holder Manifest, onMachine []Manifest, machine string) []State
|
||||
return out
|
||||
}
|
||||
|
||||
// itemData is an item's attributes in the shape of a contribution's data: each at the top and among the options, where
|
||||
// a shape's Reads looks for what a piece shows (barBlockReads: shows, options.state).
|
||||
func itemData(it Item) map[string]any {
|
||||
data, options := map[string]any{}, map[string]any{}
|
||||
for k, v := range it.Attributes {
|
||||
data[k], options[k] = v, v
|
||||
}
|
||||
data["options"] = options
|
||||
return data
|
||||
}
|
||||
|
||||
// wholeIn is v as a whole number within [lo, hi]; JSON gives every number as a float.
|
||||
func wholeIn(v any, lo, hi int) (int, bool) {
|
||||
f, ok := v.(float64)
|
||||
|
||||
@@ -394,6 +394,16 @@ func TestAStateBlockIsRenderedForThisMachineAndTheHolderIsGrantedToReadIt(t *tes
|
||||
if got := ReadsGranted(barHolder(), []Manifest{barHolder()}, "laptop"); len(got) != 0 {
|
||||
t.Fatalf("a holder alone is granted %v", got)
|
||||
}
|
||||
// Issue 774 (design 57 L3): the block as an item grants the same read, at this machine's own key alone, even where
|
||||
// the item names another machine's.
|
||||
for _, attrs := range []map[string]string{{"shows": "state", "state": "power.draw"}, {"shows": "state", "state": "power.draw", "key": "other"}} {
|
||||
moved := Manifest{Module: "power", State: []StateDeclaration{{Name: "draw"}}, Items: []Item{{
|
||||
Collection: "contribution:node-bar:block", Name: "power-draw", IfCapability: "power-meter", Attributes: attrs}}}
|
||||
if got := ReadsGranted(barHolder(), []Manifest{barHolder(), moved}, "laptop"); len(got) != 1 ||
|
||||
got[0] != (StateRead{Read: "power.draw", Key: "laptop"}) {
|
||||
t.Errorf("the holder is granted %v by the item %v", got, attrs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAModuleOffersItsOwnStateOnly(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// A seat verb a module invokes is one some holder serves (issue 774, sweep item 1). messenger invoked
|
||||
// `seat:node-engine.root-free`, which nothing in the mesh answered, so every ask that offers an approval was held on
|
||||
// Telegram, silently. A manifest names what it calls (novox/hq ADR 0152); that a seat by that name exists, promises
|
||||
// that verb, and has a holder is only known over the whole catalogue, so it is judged here, beside the seats' other
|
||||
// rules (CatalogueProblems).
|
||||
|
||||
// UnheldSeat ends the problem of a `seat:<seat>.<verb>` whose seat no module given claims: over the whole catalogue a
|
||||
// refusal, over some manifests alone a seat whose holder was not given (IsUndeclaredSeat says both).
|
||||
const UnheldSeat = "which no module claims, so nothing answers it"
|
||||
|
||||
// SeatInvoke splits an invoke `seat:<seat>.<verb>` into its seat and verb; ok is false for anything else (a module's
|
||||
// tool, `*`).
|
||||
func SeatInvoke(invoke string) (seat, verb string, ok bool) {
|
||||
rest, isSeat := strings.CutPrefix(invoke, "seat:")
|
||||
if !isSeat {
|
||||
return "", "", false
|
||||
}
|
||||
seat, verb, ok = strings.Cut(rest, ".")
|
||||
return seat, verb, ok && seat != "" && verb != ""
|
||||
}
|
||||
|
||||
// seatInvokeProblems are the `seat:<seat>.<verb>` a module or one of its processes invokes that nothing answers: a seat
|
||||
// no module declares and the mesh does not define, a verb the seat does not promise, or a seat with no holder (a
|
||||
// seat a module declares and no module claims; a mesh seat's holder, the mesh itself or a module of any repository,
|
||||
// is not judged).
|
||||
func seatInvokeProblems(shelf Shelf, declared map[string]SeatDeclaration) []string {
|
||||
claimed := map[string]bool{}
|
||||
for _, m := range shelf {
|
||||
for _, c := range m.Claims {
|
||||
claimed[c.Name] = true
|
||||
if s, ok := SeatNamed(c.Name); ok {
|
||||
claimed[s.Name] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
var problems []string
|
||||
for _, module := range shelfOrder(shelf) {
|
||||
m := shelf[module]
|
||||
for _, invoke := range invokesOf(m) {
|
||||
seat, verb, ok := SeatInvoke(invoke)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var verbs []Verb
|
||||
held := claimed[seat]
|
||||
if s, isMesh := SeatNamed(seat); isMesh {
|
||||
// A mesh seat is held by the mesh itself or by a module of any repository: one repository's catalogue
|
||||
// cannot say it has no holder, so only its verbs are judged (the lead, 2026-10-30: mesh-control's check
|
||||
// runs over the captured mesh-catalog alone, whose mesh-self-check calls the watchman's seat).
|
||||
verbs, held = s.Serves, true
|
||||
} else if d, isModule := declared[seat]; isModule {
|
||||
verbs = d.Serves
|
||||
} else {
|
||||
problems = append(problems, fmt.Sprintf("%s invokes %s, a verb of the seat %q, "+UndeclaredSeat, module, invoke, seat))
|
||||
continue
|
||||
}
|
||||
if !slices.ContainsFunc(verbs, func(v Verb) bool { return v.Name == verb }) {
|
||||
problems = append(problems, fmt.Sprintf("%s invokes %s, and the seat %s serves no verb %q", module, invoke, seat, verb))
|
||||
continue
|
||||
}
|
||||
if !held {
|
||||
problems = append(problems, fmt.Sprintf("%s invokes %s, a verb of the seat %q, "+UnheldSeat, module, invoke, seat))
|
||||
}
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// invokesOf is everything a module says it invokes: its own and each process's bus, once each.
|
||||
func invokesOf(m Manifest) []string {
|
||||
out := append([]string(nil), m.Invokes...)
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) != "process" {
|
||||
continue
|
||||
}
|
||||
if _, present := r["bus"]; !present {
|
||||
continue
|
||||
}
|
||||
if b, _, err := processBusOf(r); err == nil {
|
||||
out = append(out, b.Invokes...)
|
||||
}
|
||||
}
|
||||
slices.Sort(out)
|
||||
return slices.Compact(out)
|
||||
}
|
||||
@@ -272,7 +272,9 @@ type Shelf map[string]Manifest
|
||||
const UndeclaredSeat = "which no module declares and the mesh does not define"
|
||||
|
||||
// IsUndeclaredSeat says whether a problem CatalogueProblems gave is a seat no manifest given declares.
|
||||
func IsUndeclaredSeat(problem string) bool { return strings.HasSuffix(problem, UndeclaredSeat) }
|
||||
func IsUndeclaredSeat(problem string) bool {
|
||||
return strings.HasSuffix(problem, UndeclaredSeat) || strings.HasSuffix(problem, UnheldSeat)
|
||||
}
|
||||
|
||||
func CatalogueProblems(shelf Shelf) []string {
|
||||
var problems []string
|
||||
@@ -349,6 +351,9 @@ func CatalogueProblems(shelf Shelf) []string {
|
||||
}
|
||||
}
|
||||
|
||||
// A seat verb a module invokes is one some holder serves (issue 774).
|
||||
problems = append(problems, seatInvokeProblems(shelf, declared)...)
|
||||
|
||||
// Which seats name each path (novox/hq ADR 0287): a module's seat naming one another seat names loses.
|
||||
named, pathProblems := pathSeats(shelfOrder(shelf), shelf, declaredBy)
|
||||
problems = append(problems, pathProblems...)
|
||||
|
||||
@@ -51,12 +51,43 @@ func TestTwoModulesCannotDeclareTheSameSeat(t *testing.T) {
|
||||
}
|
||||
|
||||
// Where ADR 0110's guarantee lands under a derived set: a typo is refused, not resolved to
|
||||
// nothing at runtime.
|
||||
// nothing at runtime. And a seat verb a module invokes is one a holder serves (issue 774, sweep item 1: messenger
|
||||
// invoked a verb nothing answered, and Telegram held every ask silently).
|
||||
func TestUsingASeatNobodyDeclaresIsRefused(t *testing.T) {
|
||||
shop := Manifest{Module: "shop", Uses: []string{"telegram-sendr"}}
|
||||
got := problemsFor(t, Shelf{"telegram": telegram(), "shop": shop})
|
||||
if !strings.Contains(got, "telegram-sendr") || !strings.Contains(got, "no module declares") {
|
||||
t.Fatalf("a misspelled seat was accepted: %s", got)
|
||||
unclaimed := telegram()
|
||||
unclaimed.Claims = nil
|
||||
for _, row := range []struct {
|
||||
name, want string
|
||||
shop Manifest
|
||||
seat Manifest
|
||||
note bool // over some manifests alone it is a note (IsUndeclaredSeat)
|
||||
}{
|
||||
{"a misspelled seat used", `uses the seat "telegram-sendr", ` + UndeclaredSeat,
|
||||
Manifest{Module: "shop", Uses: []string{"telegram-sendr"}}, telegram(), true},
|
||||
{"a declared verb invoked of a held seat", "", Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.status"}}, telegram(), false},
|
||||
{"a verb the seat does not promise", `the seat telegram-sender serves no verb "resend"`,
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.resend"}}, telegram(), false},
|
||||
{"a verb of a misspelled seat", `a verb of the seat "telegram-sendr", ` + UndeclaredSeat,
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:telegram-sendr.status"}}, telegram(), true},
|
||||
{"a verb of a declared seat nobody claims", `a verb of the seat "telegram-sender", ` + UnheldSeat,
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:telegram-sender.status"}}, unclaimed, true},
|
||||
{"a verb of a mesh seat whose holder lives in another repository", "",
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:mesh-watchdogs.signals"}}, telegram(), false},
|
||||
{"a verb a mesh seat does not promise", `the seat mesh-controller serves no verb "root-freed"`,
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:mesh-controller.root-freed"}}, telegram(), false},
|
||||
{"the controller's verb: the controller answers its seat", "",
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:mesh-controller.root-free"}}, telegram(), false},
|
||||
{"the engine's verb: every node's engine holds its seat", "",
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:node-engine.root-free"}}, telegram(), false},
|
||||
{"a verb the engine's seat does not promise", `the seat node-engine serves no verb "root-freed"`,
|
||||
Manifest{Module: "shop", Invokes: []string{"seat:node-engine.root-freed"}}, telegram(), false},
|
||||
} {
|
||||
got := CatalogueProblems(Shelf{"telegram": row.seat, "shop": row.shop})
|
||||
joined := strings.Join(got, "; ")
|
||||
if row.want == "" && joined != "" || row.want != "" && (len(got) != 1 || !strings.Contains(joined, row.want) ||
|
||||
IsUndeclaredSeat(got[0]) != row.note) {
|
||||
t.Errorf("%s: %s", row.name, joined)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -59,6 +59,10 @@ func sealedFor(m Manifest, needs []Needed, with Rendering) (map[string]string, e
|
||||
for name := range m.OwnSecrets.Plain() {
|
||||
if value := with.Needed[m.Module][name]; value != "" {
|
||||
sealed[name] = value
|
||||
} else if with.Foreseen[m.Module][name] {
|
||||
// A stand-in where its own file has one (the Foreseeing and Planning modes, issue 774): a file naming it
|
||||
// composes as its own file does, never refused for a value no such composition reads.
|
||||
sealed[name] = ForeseenSealed
|
||||
}
|
||||
}
|
||||
for _, to := range m.SecretRequirements() {
|
||||
|
||||
@@ -21,34 +21,42 @@ func fileNamed(out []map[string]any, id string) map[string]any {
|
||||
return nil
|
||||
}
|
||||
|
||||
// A module's own credential, put into a file it wrote.
|
||||
// A module's own credential, put into a file it wrote: the value sealed to the node, or, in a composition asked ahead of
|
||||
// the send or for a plan alone, the stand-in its own file has (issue 774: baserow's config.json refused ace's plans).
|
||||
func TestAFileGetsTheSecretItsContentAsksFor(t *testing.T) {
|
||||
r := Resolution{Node: "anchor", Modules: []Manifest{{
|
||||
Module: "gitea",
|
||||
OwnSecrets: OwnSecrets{"admin": {Path: "/var/lib/gitea/admin.env"}},
|
||||
Resources: []map[string]any{{
|
||||
"id": "conf", "type": "file", "path": "/etc/gitea/app.ini",
|
||||
"content": "[security]\nSECRET_KEY = ${secret:admin}\n",
|
||||
}},
|
||||
}}}
|
||||
out, err := r.Declaration(Rendering{
|
||||
Needed: map[string]map[string]string{"gitea": {"admin": "sealed-admin"}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
file := fileNamed(out, "gitea.conf")
|
||||
if file == nil {
|
||||
t.Fatalf("no file in %v", out)
|
||||
}
|
||||
got, _ := file["secrets"].(map[string]any)
|
||||
if got["admin"] != "sealed-admin" {
|
||||
t.Errorf("the file was sent without its secret: %v", file)
|
||||
}
|
||||
// The hole is still a hole on the wire. The mesh does not fill it in; the host does, on the
|
||||
// machine, which is the only place both halves exist.
|
||||
if !strings.Contains(file["content"].(string), "${secret:admin}") {
|
||||
t.Errorf("the mesh substituted it itself: %v", file["content"])
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
with Rendering
|
||||
want string
|
||||
}{
|
||||
{"sealed", Rendering{Needed: map[string]map[string]string{"gitea": {"admin": "sealed-admin"}}}, "sealed-admin"},
|
||||
{"foreseen", Rendering{Foreseen: map[string]map[string]bool{"gitea": {"admin": true}}}, ForeseenSealed},
|
||||
} {
|
||||
r := Resolution{Node: "anchor", Modules: []Manifest{{
|
||||
Module: "gitea",
|
||||
OwnSecrets: OwnSecrets{"admin": {Path: "/var/lib/gitea/admin.env"}},
|
||||
Resources: []map[string]any{{
|
||||
"id": "conf", "type": "file", "path": "/etc/gitea/app.ini",
|
||||
"content": "[security]\nSECRET_KEY = ${secret:admin}\n",
|
||||
}},
|
||||
}}}
|
||||
out, err := r.Declaration(row.with)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", row.name, err)
|
||||
}
|
||||
file := fileNamed(out, "gitea.conf")
|
||||
if file == nil {
|
||||
t.Fatalf("%s: no file in %v", row.name, out)
|
||||
}
|
||||
got, _ := file["secrets"].(map[string]any)
|
||||
if got["admin"] != row.want {
|
||||
t.Errorf("%s: the file was sent without its secret: %v", row.name, file)
|
||||
}
|
||||
// The hole is still a hole on the wire. The mesh does not fill it in; the host does, on the
|
||||
// machine, which is the only place both halves exist.
|
||||
if !strings.Contains(file["content"].(string), "${secret:admin}") {
|
||||
t.Errorf("%s: the mesh substituted it itself: %v", row.name, file["content"])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/kvlimit"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -190,7 +191,7 @@ func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s: %w", bucket, err)
|
||||
}
|
||||
if _, err := composed.Put(ctx, key, encoded); err != nil {
|
||||
if _, err := kvlimit.Put(ctx, composed, key, encoded); err != nil {
|
||||
return fmt.Errorf("%s in %s: %w", key, bucket, err)
|
||||
}
|
||||
value, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digest, Judgement: delivery.JudgedCannot,
|
||||
@@ -202,7 +203,7 @@ func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s: %w", delivery.ControllerComposedBucket, err)
|
||||
}
|
||||
if _, err := digests.Put(ctx, key, value); err != nil {
|
||||
if _, err := kvlimit.Put(ctx, digests, key, value); err != nil {
|
||||
return fmt.Errorf("%s in %s: %w", key, delivery.ControllerComposedBucket, err)
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -439,6 +439,68 @@ func Plans(ctx context.Context, r Reader, mesh Mesh, s Scope) (map[string]catalo
|
||||
return PlansOf(in, s.Node)
|
||||
}
|
||||
|
||||
// PlansHeld is Plans composed module by module (issue 774: one bad module must not drop its node): a module placed on
|
||||
// the scope's node whose build cannot be read or taken in, or whose plan refuses the node's, is held, by the one rule
|
||||
// Compose follows (composeEach, culprit): composed as its stub so what it provides and requires still shapes the rest,
|
||||
// named in held with why, and none of its plan in plans. Only a refusal that is no single module's refuses the node,
|
||||
// err. The seat mesh-provisioner's holder composes so: a node with one bad manifest still has every other module's plan.
|
||||
func PlansHeld(ctx context.Context, r Reader, mesh Mesh, s Scope) (plans map[string]catalogue.Plan, held map[string]string, err error) {
|
||||
held = map[string]string{}
|
||||
in, err := input(ctx, r, mesh, s, func(module string, err error) { held[module] = err.Error() })
|
||||
if err != nil {
|
||||
return nil, held, err
|
||||
}
|
||||
plans, err = plansHeldOf(in, s.Node, held)
|
||||
return plans, held, err
|
||||
}
|
||||
|
||||
// plansHeldOf is the plans of node over in, every module in held held, and each module a refusal is of held too until
|
||||
// the plans compose (culprit's rule over the Planning mode: its stub first, then its bare stub).
|
||||
func plansHeldOf(in compose.Input, node string, held map[string]string) (map[string]catalogue.Plan, error) {
|
||||
for _, m := range sortedKeys(held) {
|
||||
in = stubbed(in, node, m, "")
|
||||
}
|
||||
compose1 := func(in compose.Input) (map[string]catalogue.Plan, error) { return plansIn(in, node, compose.Planning) }
|
||||
// Each try holds one module more or leaves out all of one held module's offer: bounded by twice the modules placed.
|
||||
for tries := 0; tries <= 2*len(assignedOn(in, node))+1; tries++ {
|
||||
plans, err := compose1(in)
|
||||
if err == nil {
|
||||
for m := range held {
|
||||
delete(plans, m)
|
||||
}
|
||||
return plans, nil
|
||||
}
|
||||
var order []string
|
||||
for _, m := range assignedOn(in, node) {
|
||||
if strings.Contains(err.Error(), m) && len(order) < maxCulprits {
|
||||
order = append(order, m)
|
||||
}
|
||||
}
|
||||
module, leave := "", ""
|
||||
find:
|
||||
for _, l := range []string{"", leaveAll} {
|
||||
for _, m := range order {
|
||||
if _, done := held[m]; done && l == "" {
|
||||
continue
|
||||
}
|
||||
if _, again := compose1(stubbed(in, node, m, l)); again == nil || again.Error() != err.Error() {
|
||||
module, leave = m, l
|
||||
break find
|
||||
}
|
||||
}
|
||||
}
|
||||
if module == "" {
|
||||
return nil, err
|
||||
}
|
||||
if _, done := held[module]; !done {
|
||||
held[module] = err.Error()
|
||||
}
|
||||
in = stubbed(in, node, module, leave)
|
||||
}
|
||||
return nil, fmt.Errorf("frombus: the plans of %s do not compose with %d module(s) held: %s", node, len(held),
|
||||
strings.Join(sortedKeys(held), ", "))
|
||||
}
|
||||
|
||||
// PlansOf is Plans over an input already filled, rendered in the Planning mode (compose.Planning): every own secret the
|
||||
// input holds none for stands in, so no value and no record of one is needed.
|
||||
func PlansOf(in compose.Input, node string) (map[string]catalogue.Plan, error) {
|
||||
|
||||
@@ -2,25 +2,75 @@ package frombustest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// ADR 0570 clause 2 (issue 1954; a3's H5): the in-memory mesh composes a plan from the bus that hands restic to the
|
||||
// provisioner and names its postgres-database, provided by postgres on n1.
|
||||
// provisioner and names its postgres-database, provided by postgres on n1. And one bad module must not drop its node
|
||||
// (issue 774): with a module placed beside it whose built manifest cannot be used, or whose plan refuses, PlansHeld
|
||||
// still hands restic's plan and names the bad one with why, where Plans refuses the whole node.
|
||||
func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
|
||||
b, err := Provisioned()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
plans, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
restic, handed := plans["restic"]
|
||||
if !handed || len(restic.Provisions) != 1 || restic.Provisions[0].Provision != "postgres-database" ||
|
||||
restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" {
|
||||
t.Fatalf("the plans are %+v", plans)
|
||||
fp := "src2:" + strings.Repeat("9", 64)
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
bad string // the indexer's manifest of a module "backup" placed on n1; empty places none
|
||||
// built is the manifest backup's done build names; empty gives it no build
|
||||
built string
|
||||
why string // words backup's hold says
|
||||
}{
|
||||
{name: "every module composes"},
|
||||
{name: "a built manifest that cannot be used",
|
||||
bad: `{"module":"backup","version":"1","build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`,
|
||||
built: `{"module":"backup","version":"1","requires":["amqp"]}`, why: "backup"},
|
||||
{name: "a plan that refuses",
|
||||
bad: `{"module":"backup","version":"1","resources":[{"id":"conf","type":"file","path":"/etc/backup.conf",` +
|
||||
`"mode":"0600","content":"key=${secret:nope}"}]}`, why: "${secret:nope}"},
|
||||
} {
|
||||
b, err := Provisioned()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if row.bad != "" {
|
||||
p := b.Placement[Node]
|
||||
p.Modules = append(p.Modules, delivery.PlacedModule{Module: "backup"})
|
||||
b.Placement[Node] = p
|
||||
digest := delivery.ManifestDigest([]byte(row.bad))
|
||||
b.Manifests[digest] = []byte(row.bad)
|
||||
b.Module["backup"] = delivery.Module{FP2: fp, Manifest: digest}
|
||||
if row.built != "" {
|
||||
builtDigest := delivery.ManifestDigest([]byte(row.built))
|
||||
b.Manifests[builtDigest] = []byte(row.built)
|
||||
b.Builds["backup "+fp] = &delivery.BuilderBuild{State: delivery.BuildDone, Manifest: builtDigest,
|
||||
Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://backup/files/blobs/x", Digest: "sha256:x"}}}
|
||||
}
|
||||
if _, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}); err == nil {
|
||||
t.Errorf("%s: Plans composed the node; the row does not refuse it", row.name)
|
||||
}
|
||||
}
|
||||
plans, held, err := frombus.PlansHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", row.name, err)
|
||||
}
|
||||
restic, handed := plans["restic"]
|
||||
if !handed || len(restic.Provisions) != 1 || restic.Provisions[0].Provision != "postgres-database" ||
|
||||
restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" {
|
||||
t.Fatalf("%s: the plans are %+v (held %v)", row.name, plans, held)
|
||||
}
|
||||
if row.bad == "" {
|
||||
if len(held) > 0 {
|
||||
t.Errorf("%s: held %v", row.name, held)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if why, ok := held["backup"]; !ok || !strings.Contains(why, row.why) || len(held) != 1 {
|
||||
t.Errorf("%s: held %v, want backup alone, saying %q", row.name, held, row.why)
|
||||
}
|
||||
if _, planned := plans["backup"]; planned {
|
||||
t.Errorf("%s: the held module has a plan", row.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+135
@@ -0,0 +1,135 @@
|
||||
package stdio
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A watch's handler runs beside the runtime, never in its way (issue 774, the sweep of 2026-10-30).
|
||||
//
|
||||
// The runtime hands a watch's current values over one at a time and gives up on the watch when none was handed over
|
||||
// within its bound (mesh-tools node-tools internal/bus/state.go StateWatch, 10 s). The bundle answered each change only
|
||||
// once its handler returned, so a slow handler (mesh-provisioner's Rejudge waits on the lock its composition holds)
|
||||
// counted against the bus's bound: the watch was given up, the SDK forgot its id, and the values the runtime still
|
||||
// delivered to it were refused as "names no watch of this bundle", 1257 times in an hour on the control-node.
|
||||
//
|
||||
// So each watch has a follower: a change is taken into it in the order the runtime hands it over and answered at once,
|
||||
// and the handler runs on the follower's own goroutine, in that order. Watch still returns only once every current
|
||||
// value was handled. A handler's error is said on the bundle's standard error, naming the watch and the change, since
|
||||
// the runtime no longer hears it. A log watch keeps its answer after the handler: its runtime sends the next entry only
|
||||
// after the answer, which is the back-pressure a replay of a long log needs.
|
||||
|
||||
// follower runs one watch's handler on its own goroutine, in order.
|
||||
type follower[T any] struct {
|
||||
what string
|
||||
handle func(T) error
|
||||
describe func(T) string
|
||||
mu sync.Mutex
|
||||
items []followed[T]
|
||||
wake chan struct{}
|
||||
}
|
||||
|
||||
// followed is one change, or with no change a mark that closes done once everything before it was handled.
|
||||
type followed[T any] struct {
|
||||
change T
|
||||
mark bool
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func newFollower[T any](what string, handle func(T) error, describe func(T) string) *follower[T] {
|
||||
f := &follower[T]{what: what, handle: handle, describe: describe, wake: make(chan struct{}, 1)}
|
||||
go f.run()
|
||||
return f
|
||||
}
|
||||
|
||||
// take queues a change; it never waits on the handler.
|
||||
func (f *follower[T]) take(change T) {
|
||||
f.mu.Lock()
|
||||
f.items = append(f.items, followed[T]{change: change})
|
||||
f.mu.Unlock()
|
||||
f.poke()
|
||||
}
|
||||
|
||||
// handled is closed once every change taken before it was handled.
|
||||
func (f *follower[T]) handled() <-chan struct{} {
|
||||
done := make(chan struct{})
|
||||
f.mu.Lock()
|
||||
f.items = append(f.items, followed[T]{mark: true, done: done})
|
||||
f.mu.Unlock()
|
||||
f.poke()
|
||||
return done
|
||||
}
|
||||
|
||||
func (f *follower[T]) poke() {
|
||||
select {
|
||||
case f.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (f *follower[T]) run() {
|
||||
for range f.wake {
|
||||
for {
|
||||
f.mu.Lock()
|
||||
if len(f.items) == 0 {
|
||||
f.mu.Unlock()
|
||||
break
|
||||
}
|
||||
next := f.items[0]
|
||||
f.items = f.items[1:]
|
||||
f.mu.Unlock()
|
||||
if next.mark {
|
||||
close(next.done)
|
||||
continue
|
||||
}
|
||||
if err := f.handle(next.change); err != nil {
|
||||
sayf("%s: the handler did not take %s: %v; the next change hands it again", f.what, f.describe(next.change), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sayf says something the runtime cannot say for the bundle, on its standard error, which its journal keeps.
|
||||
func sayf(format string, args ...any) {
|
||||
fmt.Fprintf(os.Stderr, "[mesh-sdk] "+format+"\n", args...)
|
||||
}
|
||||
|
||||
// gaveUpMost bounds what a bundle remembers of the watches it gave up.
|
||||
const gaveUpMost = 256
|
||||
|
||||
// gaveUp keeps why a watch was given up, so a change the runtime still hands to it is refused saying so, never as a
|
||||
// watch nobody asked for. Called with c.mu held.
|
||||
func (c *channel) gaveUp(id, why string) {
|
||||
if c.givenUp == nil || len(c.givenUp) >= gaveUpMost {
|
||||
c.givenUp = map[string]string{}
|
||||
}
|
||||
c.givenUp[id] = why
|
||||
}
|
||||
|
||||
// noWatch is the refusal of a change for watch id, which this bundle never asked or gave up.
|
||||
func (c *channel) noWatch(kind, id string) string {
|
||||
c.mu.Lock()
|
||||
why, given := c.givenUp[id]
|
||||
c.mu.Unlock()
|
||||
if given {
|
||||
sayf("%s: the runtime still hands over changes for the watch %s, which this bundle gave up: %s", kind, id, why)
|
||||
return fmt.Sprintf("%s: this bundle gave the watch %s up (%s); stop handing it over", kind, id, why)
|
||||
}
|
||||
return kind + " names no watch of this bundle"
|
||||
}
|
||||
|
||||
// Follow watches key of the state as Watch does and asks again until the watch is taken, saying each give-up and
|
||||
// when it asks again (2 s, doubled to a minute): the loop every module wrote for itself, in one place. It returns once
|
||||
// the watch is taken; call it beside the code that needs it.
|
||||
func (s KeptState) Follow(key string, handle func(StateChange) error) {
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
err := s.Watch(key, handle)
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
sayf("the state %s is not watched yet (%v); asking again in %s", s.name, err, wait)
|
||||
time.Sleep(wait)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,175 @@
|
||||
// Package kvlimit refuses a key-value write whose value is larger than its bucket takes, before it is put, naming the
|
||||
// bucket, the key, the size and the limit (issue 774, sweep item 5). The bus refuses such a write too, but only after
|
||||
// the put, as "message size exceeds maximum" (JetStream error 10054); a writer that retries on errors then loops on a
|
||||
// value that can never land. A refusal from here, or the bus's own one passed through Refused, is a *TooLarge that
|
||||
// errors.Is ErrTooLarge: a writer stops on it and says it, it never writes the same value again.
|
||||
//
|
||||
// A value that may outgrow its bucket is shaped one of two ways, never retried:
|
||||
// - split over keys: one key per part that changes on its own (per node, per module), each well under the limit;
|
||||
// - or kept in the object store, with only its name and digest in the key.
|
||||
package kvlimit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// ModuleStateValueBytes is the most a value in a module's declared state may hold: every module bucket is made with
|
||||
// it (buscompose.StateMaxValueBytes).
|
||||
const ModuleStateValueBytes = 256 * 1024
|
||||
|
||||
// HeaderRoom is what a write's headers may take of the limit: the bus counts headers and value together, and a create,
|
||||
// an update or a key's time to live adds one header each. A value closer to the limit than this is refused.
|
||||
const HeaderRoom = 128
|
||||
|
||||
// codeMessageExceedsMaximum is JetStream's error code for a message over the stream's largest (JSStreamMessageExceedsMaximumErr).
|
||||
const codeMessageExceedsMaximum jetstream.ErrorCode = 10054
|
||||
|
||||
// ErrTooLarge is what every *TooLarge is: a value its bucket does not take, which no retry will change.
|
||||
var ErrTooLarge = errors.New("a value larger than its bucket takes")
|
||||
|
||||
// TooLarge is one refused write. Limit is zero when the bus refused it and its limit was not known here.
|
||||
type TooLarge struct {
|
||||
Bucket, Key string
|
||||
Size, Limit int64
|
||||
}
|
||||
|
||||
func (e *TooLarge) Error() string {
|
||||
limit := "the most its bucket takes"
|
||||
if e.Limit > 0 {
|
||||
limit = fmt.Sprintf("the %d bytes the bucket takes (headers included)", e.Limit)
|
||||
}
|
||||
return fmt.Sprintf("the value for %s in %s is %d bytes, over %s: it is refused and written again it is refused again; "+
|
||||
"split it over keys or keep it in the object store with its name in the key", e.Key, e.Bucket, e.Size, limit)
|
||||
}
|
||||
|
||||
func (e *TooLarge) Unwrap() error { return ErrTooLarge }
|
||||
|
||||
// Check refuses a value of size bytes for key in bucket when it does not fit limit with HeaderRoom to spare; a limit of
|
||||
// zero or less is no limit.
|
||||
func Check(bucket, key string, size int, limit int64) error {
|
||||
if limit > 0 && int64(size)+HeaderRoom > limit {
|
||||
return &TooLarge{Bucket: bucket, Key: key, Size: int64(size), Limit: limit}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Refused is err as a *TooLarge when it is the bus's refusal of a value too large (the stream's 10054, the client's
|
||||
// maximum payload, or either said in words by a runtime that passed it on), and err unchanged otherwise.
|
||||
func Refused(bucket, key string, size int, limit int64, err error) error {
|
||||
if err == nil || errors.Is(err, ErrTooLarge) {
|
||||
return err
|
||||
}
|
||||
var api *jetstream.APIError
|
||||
tooLarge := errors.As(err, &api) && api.ErrorCode == codeMessageExceedsMaximum
|
||||
tooLarge = tooLarge || errors.Is(err, nats.ErrMaxPayload)
|
||||
words := err.Error()
|
||||
tooLarge = tooLarge || strings.Contains(words, "message size exceeds maximum") || strings.Contains(words, "maximum payload exceeded")
|
||||
if !tooLarge {
|
||||
return err
|
||||
}
|
||||
forget(bucket)
|
||||
return fmt.Errorf("%w (the bus: %v)", &TooLarge{Bucket: bucket, Key: key, Size: int64(size), Limit: limit}, err)
|
||||
}
|
||||
|
||||
// limits are the buckets' largest messages as read, by bucket name: a limit is kept once read, and only a limit (an
|
||||
// unlimited bucket is read again each time). A kept limit that refuses a value is read again before the refusal is
|
||||
// given, so a bucket made again with a larger limit is never refused by an old one; one made with a smaller limit is
|
||||
// refused by the bus, and Refused forgets the kept one.
|
||||
var limits sync.Map // bucket name → int64
|
||||
|
||||
func forget(bucket string) { limits.Delete(bucket) }
|
||||
|
||||
// LimitOf is the largest message kv's bucket takes, read from its stream and kept; zero when it has no limit or it
|
||||
// cannot be read (the write then goes, and Refused names the bus's refusal).
|
||||
func LimitOf(ctx context.Context, kv jetstream.KeyValue) int64 {
|
||||
if l, ok := limits.Load(nameOf(kv)); ok {
|
||||
return l.(int64)
|
||||
}
|
||||
return readLimit(ctx, kv)
|
||||
}
|
||||
|
||||
func readLimit(ctx context.Context, kv jetstream.KeyValue) (limit int64) {
|
||||
name := nameOf(kv)
|
||||
if name == "" {
|
||||
return 0
|
||||
}
|
||||
// A test's fake that implements only the writes it needs panics on Status; its limit is not known.
|
||||
defer func() {
|
||||
if recover() != nil {
|
||||
limit = 0
|
||||
}
|
||||
}()
|
||||
st, err := kv.Status(ctx)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
s, ok := st.(*jetstream.KeyValueBucketStatus)
|
||||
if !ok || s.StreamInfo() == nil || s.StreamInfo().Config.MaxMsgSize <= 0 {
|
||||
forget(name)
|
||||
return 0
|
||||
}
|
||||
limit = int64(s.StreamInfo().Config.MaxMsgSize)
|
||||
limits.Store(name, limit)
|
||||
return limit
|
||||
}
|
||||
|
||||
// fits is the limit of kv's bucket and nil when a value of size bytes fits it, or the *TooLarge refusing it, checked
|
||||
// against a freshly read limit.
|
||||
func fits(ctx context.Context, kv jetstream.KeyValue, key string, size int) (int64, error) {
|
||||
limit := LimitOf(ctx, kv)
|
||||
if Check(nameOf(kv), key, size, limit) == nil {
|
||||
return limit, nil
|
||||
}
|
||||
limit = readLimit(ctx, kv)
|
||||
return limit, Check(nameOf(kv), key, size, limit)
|
||||
}
|
||||
|
||||
// nameOf is kv's bucket name, or "" for none or a test's fake that does not say it.
|
||||
func nameOf(kv jetstream.KeyValue) (name string) {
|
||||
if kv == nil {
|
||||
return ""
|
||||
}
|
||||
defer func() {
|
||||
if recover() != nil {
|
||||
name = ""
|
||||
}
|
||||
}()
|
||||
return kv.Bucket()
|
||||
}
|
||||
|
||||
// Put is kv.Put, refused before it is put when value does not fit the bucket.
|
||||
func Put(ctx context.Context, kv jetstream.KeyValue, key string, value []byte) (uint64, error) {
|
||||
limit, err := fits(ctx, kv, key, len(value))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Put(ctx, key, value)
|
||||
return rev, Refused(nameOf(kv), key, len(value), limit, err)
|
||||
}
|
||||
|
||||
// Create is kv.Create, refused before it is put when value does not fit the bucket.
|
||||
func Create(ctx context.Context, kv jetstream.KeyValue, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) {
|
||||
limit, err := fits(ctx, kv, key, len(value))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Create(ctx, key, value, opts...)
|
||||
return rev, Refused(nameOf(kv), key, len(value), limit, err)
|
||||
}
|
||||
|
||||
// Update is kv.Update, refused before it is put when value does not fit the bucket.
|
||||
func Update(ctx context.Context, kv jetstream.KeyValue, key string, value []byte, revision uint64) (uint64, error) {
|
||||
limit, err := fits(ctx, kv, key, len(value))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Update(ctx, key, value, revision)
|
||||
return rev, Refused(nameOf(kv), key, len(value), limit, err)
|
||||
}
|
||||
+9
-3
@@ -6,7 +6,10 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
)
|
||||
|
||||
// Ask reads the router's record of one of this process's module's own asks, under its own module and no
|
||||
@@ -24,9 +27,12 @@ func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) {
|
||||
return nil, err
|
||||
}
|
||||
subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
reply, err := c.nc.RequestWithContext(ctx, subject, nil)
|
||||
// A read: asked again while the bus answers nothing (busretry).
|
||||
reply, err := busretry.Do("reading "+subject, func() (*nats.Msg, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
return c.nc.RequestWithContext(ctx, subject, nil)
|
||||
}, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading %s's ask %s from %s: %w", c.account.Module, id, bucket, err)
|
||||
}
|
||||
|
||||
@@ -10,6 +10,8 @@ import (
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/kvlimit"
|
||||
)
|
||||
|
||||
// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the delivery
|
||||
@@ -167,7 +169,7 @@ func (b *AwaitedBucket) Put(ctx context.Context, key string, value []byte) (uint
|
||||
kv = b.kv
|
||||
b.mu.Unlock()
|
||||
}
|
||||
return kv.Put(ctx, key, value)
|
||||
return kvlimit.Put(ctx, kv, key, value)
|
||||
}
|
||||
|
||||
// bind tries one STREAM.INFO of the bucket and, when it is there, binds it; it answers the bucket or nil.
|
||||
@@ -196,7 +198,7 @@ func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.Ke
|
||||
written, failed := 0, 0
|
||||
for _, key := range b.order {
|
||||
ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
|
||||
_, err := kv.Put(ctx, key, b.pending[key])
|
||||
_, err := kvlimit.Put(ctx, kv, key, b.pending[key])
|
||||
cancel()
|
||||
if err != nil {
|
||||
failed++
|
||||
|
||||
+35
-5
@@ -9,6 +9,8 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
)
|
||||
|
||||
// CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35,
|
||||
@@ -47,16 +49,44 @@ func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) {
|
||||
return c.callOn(called, fallback, body)
|
||||
}
|
||||
|
||||
// callOn asks called, then fallback when nothing answers it; while the bus itself answers nothing (a user-list reload
|
||||
// drops the account's imports for a few seconds, busretry) both are asked again, within busretry.Bound. A call nobody
|
||||
// answers on a bus that answers is not asked again: its callee is not running. A call that may have been delivered (a
|
||||
// timeout, any answer) is never asked again.
|
||||
func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error) {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := c.request(called, data)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
result, err = c.request(fallback, data)
|
||||
return callWith(c.request, c.blackedOut, called, fallback, data)
|
||||
}
|
||||
|
||||
// callWith is callOn over a request and a blackout probe. Only the transport's own errors can count as undelivered: a
|
||||
// callee's answer is *busretry.Answered, whatever its words, so a call that reached its callee is never asked again.
|
||||
func callWith(request func(string, []byte) (json.RawMessage, error), blackout func() bool, called, fallback string,
|
||||
data []byte) (json.RawMessage, error) {
|
||||
return busretry.Do("the call on "+fallback, func() (json.RawMessage, error) {
|
||||
result, err := request(called, data)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
result, err = request(fallback, data)
|
||||
}
|
||||
return result, err
|
||||
}, blackout)
|
||||
}
|
||||
|
||||
// answered is the error a callee answered with: delivered, never asked again (busretry.Answered).
|
||||
func answered(words string) error { return &busretry.Answered{Words: words} }
|
||||
|
||||
// blackedOut says the bus answers this account nothing at all: its JetStream API, which every account imports, has no
|
||||
// responder or says it is not enabled.
|
||||
func (c *Conn) blackedOut() bool {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
msg, err := c.nc.RequestWithContext(ctx, "$JS.API.INFO", nil)
|
||||
if err != nil {
|
||||
return busretry.Undelivered(err)
|
||||
}
|
||||
return result, err
|
||||
return strings.Contains(string(msg.Data), `"err_code":10039`) || strings.Contains(string(msg.Data), "jetstream not enabled")
|
||||
}
|
||||
|
||||
// callerToken is this account's user name as the last token of a call it publishes, each dot written ~
|
||||
@@ -116,7 +146,7 @@ func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) {
|
||||
return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err)
|
||||
}
|
||||
if r.Error != "" {
|
||||
return nil, errors.New(r.Error)
|
||||
return nil, answered(r.Error)
|
||||
}
|
||||
return r.Result, nil
|
||||
}
|
||||
|
||||
@@ -1,8 +1,14 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
)
|
||||
|
||||
// 1191/AC3: given a module tool key <module>.<tool>, when called, then the subject is
|
||||
@@ -79,3 +85,28 @@ func TestCallOnRefusesANodeThatIsNotOneToken(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The review of 2026-10-30 (issue 774): a callee's own answer is delivered, whatever its words, and is never asked
|
||||
// again during a blackout; only the transport's no responders is.
|
||||
func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) {
|
||||
busretry.Bound, busretry.FirstWait, busretry.MostWait = 50*time.Millisecond, time.Millisecond, 2*time.Millisecond
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
answer []error // per ask; the last repeats
|
||||
asks int
|
||||
}{
|
||||
{"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1},
|
||||
{"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1},
|
||||
{"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3},
|
||||
} {
|
||||
asks := 0
|
||||
_, err := callWith(func(string, []byte) (json.RawMessage, error) {
|
||||
e := row.answer[min(asks, len(row.answer)-1)]
|
||||
asks++
|
||||
return json.RawMessage(`{}`), e
|
||||
}, func() bool { return true }, "called", "fallback", nil)
|
||||
if asks != row.asks {
|
||||
t.Errorf("%s: asked %d times (%v), want %d", row.name, asks, err, row.asks)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+47
-4
@@ -13,6 +13,8 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"git.novox.be/novox/mesh-sdk/go/kvlimit"
|
||||
)
|
||||
|
||||
// stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is
|
||||
@@ -73,7 +75,12 @@ func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
|
||||
}
|
||||
|
||||
// Get is a key's current value, or nil when it has none.
|
||||
// Get waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
|
||||
func (s State) Get(key string) (*stdio.StateEntry, error) {
|
||||
return busretry.Do(s.what("get "+key), func() (*stdio.StateEntry, error) { return s.get(key) }, nil)
|
||||
}
|
||||
|
||||
func (s State) get(key string) (*stdio.StateEntry, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
@@ -92,7 +99,12 @@ func (s State) Get(key string) (*stdio.StateEntry, error) {
|
||||
|
||||
// Create writes a key only when it has none, and returns the revision; a key that has one is refused as
|
||||
// stdio.ErrChanged.
|
||||
// Create waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
|
||||
func (s State) Create(key string, value any) (uint64, error) {
|
||||
return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.create(key, value) }, nil)
|
||||
}
|
||||
|
||||
func (s State) create(key string, value any) (uint64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
@@ -103,7 +115,7 @@ func (s State) Create(key string, value any) (uint64, error) {
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Create(ctx, key, data)
|
||||
rev, err := kvlimit.Create(ctx, kv, key, data)
|
||||
if errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
||||
}
|
||||
@@ -120,7 +132,12 @@ const MinTTL = time.Second
|
||||
// 1952 S6): the bus removes it when ttl passes and its watchers hear the removal, so an expiry is an event and no timer
|
||||
// runs. A key that has a value is refused as stdio.ErrChanged: an act begun once cannot be begun again. The bucket must
|
||||
// take keys with their own time to live (a state declared `key-ttl`).
|
||||
// CreateTTL waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
|
||||
func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) {
|
||||
return busretry.Do(s.what("create "+key), func() (uint64, error) { return s.createTTL(key, value, ttl) }, nil)
|
||||
}
|
||||
|
||||
func (s State) createTTL(key string, value any, ttl time.Duration) (uint64, error) {
|
||||
if ttl < MinTTL {
|
||||
return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", s.Module, s.Name, key, ttl, MinTTL)
|
||||
}
|
||||
@@ -134,7 +151,7 @@ func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, erro
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Create(ctx, key, data, jetstream.KeyTTL(ttl))
|
||||
rev, err := kvlimit.Create(ctx, kv, key, data, jetstream.KeyTTL(ttl))
|
||||
if errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
||||
}
|
||||
@@ -160,7 +177,12 @@ func (s State) RenewTTL(key string, value any, ttl time.Duration) (uint64, error
|
||||
|
||||
// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
|
||||
// is refused as stdio.ErrChanged.
|
||||
// Update waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
|
||||
func (s State) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
return busretry.Do(s.what("update "+key), func() (uint64, error) { return s.update(key, value, revision) }, nil)
|
||||
}
|
||||
|
||||
func (s State) update(key string, value any, revision uint64) (uint64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
@@ -171,7 +193,7 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Update(ctx, key, data, revision)
|
||||
rev, err := kvlimit.Update(ctx, kv, key, data, revision)
|
||||
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
||||
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
||||
}
|
||||
@@ -182,7 +204,12 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
}
|
||||
|
||||
// Put writes a key whole, whatever it held, and returns the revision written.
|
||||
// Put waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
|
||||
func (s State) Put(key string, value any) (uint64, error) {
|
||||
return busretry.Do(s.what("put "+key), func() (uint64, error) { return s.put(key, value) }, nil)
|
||||
}
|
||||
|
||||
func (s State) put(key string, value any) (uint64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
@@ -193,7 +220,7 @@ func (s State) Put(key string, value any) (uint64, error) {
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Put(ctx, key, data)
|
||||
rev, err := kvlimit.Put(ctx, kv, key, data)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
||||
}
|
||||
@@ -201,7 +228,12 @@ func (s State) Put(key string, value any) (uint64, error) {
|
||||
}
|
||||
|
||||
// Keys is every key with a current value, sorted; none is an empty list.
|
||||
// Keys waits out a bus that answers nothing for a few seconds (busretry): only a request that reached no one is asked again.
|
||||
func (s State) Keys() ([]string, error) {
|
||||
return busretry.Do(s.what("keys"), func() ([]string, error) { return s.keys() }, nil)
|
||||
}
|
||||
|
||||
func (s State) keys() ([]string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
@@ -313,7 +345,18 @@ func (s State) Watch(key string, handle func(stdio.StateChange) error) error {
|
||||
}
|
||||
|
||||
// Delete removes a key. A key that was not there is not an error.
|
||||
// Delete waits out a bus that answers nothing for a few seconds, as Put does.
|
||||
func (s State) Delete(key string) error {
|
||||
_, err := busretry.Do(s.what("delete "+key), func() (struct{}, error) { return struct{}{}, s.delete(key) }, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
// what names a request on this state for busretry.
|
||||
func (s State) what(request string) string {
|
||||
return fmt.Sprintf("%s's state %s, %s", s.Module, s.Name, request)
|
||||
}
|
||||
|
||||
func (s State) delete(key string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/kvlimit"
|
||||
)
|
||||
|
||||
// 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a
|
||||
@@ -148,7 +149,8 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
|
||||
func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
|
||||
url, account := grantedSendProcess(t)
|
||||
_, adminJetStream := adminJS(t, url)
|
||||
if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
|
||||
admins, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends", MaxValueSize: 1024})
|
||||
if err != nil {
|
||||
t.Fatalf("the admin could not make the bucket: %v", err)
|
||||
}
|
||||
conn, err := Connect(account)
|
||||
@@ -158,6 +160,27 @@ func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
|
||||
defer conn.Close()
|
||||
s := conn.State("sends")
|
||||
|
||||
// Issue 774 (sweep item 5): a value over the bucket's largest is refused before it is put, naming the bucket, the key,
|
||||
// its size and the limit, as kvlimit.ErrTooLarge, so a writer stops on it; nothing reaches the bucket.
|
||||
big := strings.Repeat("x", 2000)
|
||||
for name, write := range map[string]func() (uint64, error){
|
||||
"put": func() (uint64, error) { return s.Put("big", big) },
|
||||
"create": func() (uint64, error) { return s.Create("big", big) },
|
||||
"update": func() (uint64, error) { return s.Update("big", big, 1) },
|
||||
} {
|
||||
_, err := write()
|
||||
var tooLarge *kvlimit.TooLarge
|
||||
if !errors.Is(err, kvlimit.ErrTooLarge) || !errors.As(err, &tooLarge) || tooLarge.Limit != 1024 ||
|
||||
tooLarge.Bucket != "mesh-mail_sends" || tooLarge.Key != "big" || tooLarge.Size != 2002 || strings.Contains(err.Error(), "the bus:") {
|
||||
t.Errorf("a %s of 2002 bytes into a bucket of 1024: %v", name, err)
|
||||
}
|
||||
}
|
||||
// The bus's own refusal (JetStream 10054), as a writer that put past the check meets it, is the same refusal.
|
||||
_, err = admins.Put(ctxFor(t), "big", []byte(big))
|
||||
if err == nil || !errors.Is(kvlimit.Refused("mesh-mail_sends", "big", 2000, 1024, err), kvlimit.ErrTooLarge) {
|
||||
t.Errorf("the bus's refusal is not ErrTooLarge: %v", err)
|
||||
}
|
||||
|
||||
if keys, err := s.Keys(); err != nil || len(keys) != 0 {
|
||||
t.Fatalf("Keys of an empty bucket: %v, %v", keys, err)
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asker"
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
)
|
||||
|
||||
// fakeEntry is a fakeKV's KeyValueEntry: only Key, Value and Revision are read by State.
|
||||
@@ -206,6 +207,19 @@ func TestStatePutWritesAKeyWhole(t *testing.T) {
|
||||
if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` {
|
||||
t.Fatalf("Get after Put: %+v, %v", e, err)
|
||||
}
|
||||
// Issue 774: a bus that answers nothing for a moment (a user-list reload) is waited out, not a lost write.
|
||||
busretry.FirstWait = time.Millisecond
|
||||
kv, blackedOut := newFakeKV(), 2
|
||||
dark := State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) {
|
||||
if blackedOut > 0 {
|
||||
blackedOut--
|
||||
return nil, jetstream.ErrJetStreamNotEnabled
|
||||
}
|
||||
return kv, nil
|
||||
}}
|
||||
if rev, err := dark.Put("k", 1); err != nil || rev == 0 || blackedOut != 0 {
|
||||
t.Fatalf("a Put through a short blackout: %d, %v", rev, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error.
|
||||
|
||||
+6
-1
@@ -3,6 +3,7 @@ package stdio
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -42,15 +43,19 @@ func provisionWatch(kind, key string, handle func(ProvisionChange) error) error
|
||||
return err
|
||||
}
|
||||
id := c.newID()
|
||||
f := newFollower("the watch "+id+" of the "+kind+" "+key, handle, func(ch ProvisionChange) string { return ch.Kind + " " + ch.Key })
|
||||
c.mu.Lock()
|
||||
c.provisioned[id] = handle
|
||||
c.provisioned[id] = f
|
||||
c.mu.Unlock()
|
||||
if _, err := c.askAs(id, "mesh/provision.watch", map[string]any{"kind": kind, "key": key}, provisionTimeout); err != nil {
|
||||
c.mu.Lock()
|
||||
delete(c.provisioned, id)
|
||||
c.gaveUp(id, fmt.Sprintf("the %s %s at %s: %v", kind, key, time.Now().Format(time.RFC3339), err))
|
||||
c.mu.Unlock()
|
||||
sayf("the watch %s of the %s %s is given up: %v", id, kind, key, err)
|
||||
return err
|
||||
}
|
||||
<-f.handled()
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
+35
-3
@@ -5,6 +5,9 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"git.novox.be/novox/mesh-sdk/go/kvlimit"
|
||||
)
|
||||
|
||||
// State — the current value of something a module keeps on the bus, which every machine sees,
|
||||
@@ -64,12 +67,33 @@ func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, err
|
||||
if err := refusedUnannounced(s.name); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// A write's value is measured before it is sent: one over the bucket's limit is refused here, by name, and a
|
||||
// refusal the bus gives anyway comes back as the same *kvlimit.TooLarge, so no writer retries it (issue 774).
|
||||
size := -1
|
||||
key, _ := params["key"].(string)
|
||||
if value, ok := params["value"]; ok {
|
||||
data, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the value for %s in the state %s: %w", key, s.name, err)
|
||||
}
|
||||
if err := kvlimit.Check("the state "+s.name, key, len(data), kvlimit.ModuleStateValueBytes); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
params["value"], size = json.RawMessage(data), len(data)
|
||||
}
|
||||
c, err := live()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
params["state"] = s.name
|
||||
return c.ask("mesh/state."+verb, params, stateTimeout)
|
||||
// A request the runtime's bus delivered to no one is asked again for a few seconds (busretry): nothing was done.
|
||||
raw, err := busretry.Do("the state "+s.name+", "+verb+" "+key, func() (json.RawMessage, error) {
|
||||
return c.ask("mesh/state."+verb, params, stateTimeout)
|
||||
}, nil)
|
||||
if size >= 0 {
|
||||
err = kvlimit.Refused("the state "+s.name, key, size, kvlimit.ModuleStateValueBytes, err)
|
||||
}
|
||||
return raw, err
|
||||
}
|
||||
|
||||
// Get is a key's current value, or nil when it has none.
|
||||
@@ -193,21 +217,29 @@ func (s KeptState) Keys() ([]string, error) {
|
||||
|
||||
// Watch hands handle every current value — none that is deleted — and then every change, in order,
|
||||
// and returns once the current values have all been handled. key narrows it: a key, with `*` for one
|
||||
// name and `**` for the rest; empty is every key. The watch lasts as long as the process.
|
||||
// name and `**` for the rest; empty is every key. The watch lasts as long as the process. handle runs beside the
|
||||
// runtime (follower.go): its time never counts against the runtime's bound on the current values, and an error it
|
||||
// returns is said on standard error. A watch the runtime refuses is given up, said, and a change the runtime still
|
||||
// hands to it is refused saying so; Follow asks again until it is taken.
|
||||
func (s KeptState) Watch(key string, handle func(StateChange) error) error {
|
||||
c, err := live()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
id := c.newID()
|
||||
f := newFollower("the watch "+id+" of the state "+s.name, handle, func(ch StateChange) string { return ch.Op + " " + ch.Key })
|
||||
c.mu.Lock()
|
||||
c.watches[id] = handle
|
||||
c.watches[id] = f
|
||||
c.mu.Unlock()
|
||||
if _, err := c.askAs(id, "mesh/state.watch", map[string]any{"state": s.name, "key": key}, stateTimeout); err != nil {
|
||||
c.mu.Lock()
|
||||
delete(c.watches, id)
|
||||
c.gaveUp(id, fmt.Sprintf("the state %s at %s: %v", s.name, time.Now().Format(time.RFC3339), err))
|
||||
c.mu.Unlock()
|
||||
sayf("the watch %s of the state %s is given up: %v", id, s.name, err)
|
||||
return err
|
||||
}
|
||||
// The runtime hands every current value over before it answers the watch: each is in the follower by now.
|
||||
<-f.handled()
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
|
||||
outR, outW := io.Pipe()
|
||||
seen := make(chan string, 16)
|
||||
watched := make(chan error, 1)
|
||||
slowWatched, release := make(chan error, 1), make(chan struct{})
|
||||
go func() {
|
||||
_ = ServeOn("", []Tool{
|
||||
{Name: "start", Run: func(map[string]any) (any, error) {
|
||||
@@ -31,6 +32,19 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
|
||||
}()
|
||||
return "started", nil
|
||||
}},
|
||||
{Name: "slow", Run: func(map[string]any) (any, error) {
|
||||
go func() {
|
||||
slowWatched <- State("slow").Watch("", func(c StateChange) error {
|
||||
<-release
|
||||
seen <- "slow " + c.Key
|
||||
return errors.New("the handler's own error")
|
||||
})
|
||||
}()
|
||||
return "started", nil
|
||||
}},
|
||||
{Name: "refused", Run: func(map[string]any) (any, error) {
|
||||
return nil, State("gone").Watch("", func(StateChange) error { return nil })
|
||||
}},
|
||||
{Name: "remember", Run: func(a map[string]any) (any, error) {
|
||||
return State("servers").Put(a["key"].(string), a["value"])
|
||||
}},
|
||||
@@ -111,6 +125,50 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
|
||||
t.Fatalf("a change for no watch was answered %v", r)
|
||||
}
|
||||
|
||||
// Issue 774 (the provisioner's "names no watch of this bundle"): a change is answered as soon as it is taken, so a
|
||||
// slow handler never counts against the runtime's bound on the current values; Watch still returns only once the
|
||||
// current values were handled, and a handler's error stays the bundle's to say.
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 60, "method": "tools/call", "params": map[string]any{"name": "slow"}})
|
||||
var slowAsk map[string]any
|
||||
for i := 0; i < 2; i++ {
|
||||
if m := next(); m["method"] == "mesh/state.watch" {
|
||||
slowAsk = m
|
||||
}
|
||||
}
|
||||
for i, key := range []string{"a", "b"} {
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 61 + i, "method": "mesh/state", "params": map[string]any{"watch": slowAsk["id"],
|
||||
"change": map[string]any{"state": "slow", "key": key, "op": "put", "value": 1, "current": true}}})
|
||||
if r := next(); r["id"] != float64(61+i) || r["error"] != nil {
|
||||
t.Fatalf("a change to a slow handler was answered %v", r)
|
||||
}
|
||||
}
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": slowAsk["id"], "result": map[string]any{}})
|
||||
select {
|
||||
case err := <-slowWatched:
|
||||
t.Fatalf("the watch returned before its current values were handled: %v", err)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
close(release)
|
||||
if a, b := <-seen, <-seen; a != "slow a" || b != "slow b" {
|
||||
t.Fatalf("handled out of order: %q, %q", a, b)
|
||||
}
|
||||
if err := <-slowWatched; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// A watch the runtime refuses is given up, and a change the runtime still hands to it is refused saying so.
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 70, "method": "tools/call", "params": map[string]any{"name": "refused"}})
|
||||
goneAsk := next()
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": goneAsk["id"], "error": map[string]any{"code": -32000,
|
||||
"message": "the current values of gone did not arrive in 10s"}})
|
||||
if r := next(); r["id"] != float64(70) {
|
||||
t.Fatalf("the refused watch answered %v", r)
|
||||
}
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 71, "method": "mesh/state", "params": map[string]any{"watch": goneAsk["id"],
|
||||
"change": map[string]any{"state": "gone", "key": "k", "op": "put"}}})
|
||||
if r := next(); r["id"] != float64(71) || !strings.Contains(fmt.Sprint(r), "gave the watch") || !strings.Contains(fmt.Sprint(r), "did not arrive in 10s") {
|
||||
t.Fatalf("a change for a given-up watch was answered %v", r)
|
||||
}
|
||||
|
||||
// Put: asked of the runtime as the module, the revision is the tool's answer.
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "remember",
|
||||
"arguments": map[string]any{"key": "all.two", "value": map[string]any{"url": "http://two"}}}})
|
||||
@@ -122,6 +180,21 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
|
||||
if r := next(); r["id"] != float64(2) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "9" {
|
||||
t.Fatalf("the put answered %v", r)
|
||||
}
|
||||
// Issue 774 (sweep item 5): a value over a module state's largest is refused before the runtime is asked, naming the
|
||||
// key, its size and the limit; one the runtime's bus refuses anyway comes back as the same refusal.
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 30, "method": "tools/call", "params": map[string]any{"name": "remember",
|
||||
"arguments": map[string]any{"key": "all.big", "value": strings.Repeat("x", 300*1024)}}})
|
||||
if r := next(); r["id"] != float64(30) || !strings.Contains(fmt.Sprint(r), "all.big in the state servers is 307202 bytes, over the 262144") {
|
||||
t.Fatalf("a value over the limit was not refused before the runtime was asked: %v", r)
|
||||
}
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 31, "method": "tools/call", "params": map[string]any{"name": "remember",
|
||||
"arguments": map[string]any{"key": "all.near", "value": "x"}}})
|
||||
near := next()
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": near["id"], "error": map[string]any{"code": -32000,
|
||||
"message": "nats: API error: code=400 err_code=10054 description=message size exceeds maximum allowed"}})
|
||||
if r := next(); r["id"] != float64(31) || !strings.Contains(fmt.Sprint(r), "written again it is refused again") {
|
||||
t.Fatalf("the bus's refusal was not named as too large: %v", r)
|
||||
}
|
||||
// PutTTL and CreateTTL ask the runtime's put and create with the key's time to live in whole seconds; a create the
|
||||
// runtime refuses as changed (-32010) is ErrChanged.
|
||||
for i, create := range []bool{false, true} {
|
||||
|
||||
+21
-28
@@ -69,9 +69,11 @@ type channel struct {
|
||||
// subscribed numbers the listeners, so one whose subscribe failed is the one removed.
|
||||
subscribed int
|
||||
// watches are this bundle's state watches by the id each was asked with (ADR 0201).
|
||||
watches map[string]func(StateChange) error
|
||||
watches map[string]*follower[StateChange]
|
||||
// provisioned are this bundle's watches of its values, by the id each was asked with (provision.go).
|
||||
provisioned map[string]func(ProvisionChange) error
|
||||
provisioned map[string]*follower[ProvisionChange]
|
||||
// givenUp are the watches this bundle gave up, by id, each with why (follower.go).
|
||||
givenUp map[string]string
|
||||
// logWatches are this bundle's watches of a log, by the id each was asked with (log.go).
|
||||
logWatches map[string]func(LogEntry) error
|
||||
// takers and provers are this bundle's seat work and proof handlers, by worker and subject (ADR 0259).
|
||||
@@ -183,8 +185,8 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
|
||||
for _, t := range tools {
|
||||
byName[t.Name] = t
|
||||
}
|
||||
c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{},
|
||||
provisioned: map[string]func(ProvisionChange) error{}, logWatches: map[string]func(LogEntry) error{},
|
||||
c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]*follower[StateChange]{},
|
||||
provisioned: map[string]*follower[ProvisionChange]{}, logWatches: map[string]func(LogEntry) error{},
|
||||
takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}}
|
||||
currentMu.Lock()
|
||||
current = c
|
||||
@@ -296,27 +298,23 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
}(id, *p.Envelope, matching)
|
||||
case "mesh/state":
|
||||
// One change to a state this bundle watches (ADR 0201), named by the watch's id, answered
|
||||
// once its handler has taken it. Beside the loop, like a tool call.
|
||||
// One change to a state this bundle watches (ADR 0201), named by the watch's id: taken into the watch's
|
||||
// follower in the order it came and answered at once, so a slow handler never counts against the runtime's
|
||||
// bound on the current values (follower.go, issue 774).
|
||||
var p struct {
|
||||
Watch string `json:"watch"`
|
||||
Change *StateChange `json:"change"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Params, &p)
|
||||
c.mu.Lock()
|
||||
handle := c.watches[p.Watch]
|
||||
f := c.watches[p.Watch]
|
||||
c.mu.Unlock()
|
||||
if handle == nil || p.Change == nil {
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/state names no watch of this bundle"}})
|
||||
if f == nil || p.Change == nil {
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": c.noWatch("mesh/state", p.Watch)}})
|
||||
continue
|
||||
}
|
||||
go func(id any, change StateChange) {
|
||||
if err := handle(change); err != nil {
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
|
||||
return
|
||||
}
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
}(id, *p.Change)
|
||||
f.take(*p.Change)
|
||||
go c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
case "mesh/log":
|
||||
// One entry of a log this bundle watches (log.go), named by the watch's id, answered once its handler has
|
||||
// taken it, in order: the runtime sends the next only after this answer.
|
||||
@@ -340,27 +338,22 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
}(id, *p.Entry)
|
||||
case "mesh/provision":
|
||||
// One change to a value of this module's that this bundle watches (provision.go), named by the
|
||||
// watch's id, answered once its handler has taken it. Beside the loop, like a tool call.
|
||||
// One change to a value of this module's that this bundle watches (provision.go), named by the watch's id:
|
||||
// taken into the watch's follower and answered at once, as a state's change is (follower.go, issue 774).
|
||||
var p struct {
|
||||
Watch string `json:"watch"`
|
||||
Change *ProvisionChange `json:"change"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Params, &p)
|
||||
c.mu.Lock()
|
||||
handle := c.provisioned[p.Watch]
|
||||
f := c.provisioned[p.Watch]
|
||||
c.mu.Unlock()
|
||||
if handle == nil || p.Change == nil {
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/provision names no watch of this bundle"}})
|
||||
if f == nil || p.Change == nil {
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": c.noWatch("mesh/provision", p.Watch)}})
|
||||
continue
|
||||
}
|
||||
go func(id any, change ProvisionChange) {
|
||||
if err := handle(change); err != nil {
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
|
||||
return
|
||||
}
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
}(id, *p.Change)
|
||||
f.take(*p.Change)
|
||||
go c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
case "mesh/work":
|
||||
// One piece of a seat's work this bundle takes (novox/hq ADR 0259): taken by an answer that is
|
||||
// not an error, offered again after one. Beside the loop, like a tool call.
|
||||
|
||||
Reference in New Issue
Block a user