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

The delivery SDK's enginereads, findings and heard leave for the clients of their seats (ADR 0634 sections 1 and 2, issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 19:22:05 +01:00
parent ddaff38e75
commit f6f18d04e1
30 changed files with 30 additions and 950 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+1 -1
View File
@@ -51,7 +51,7 @@ import (
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-delivery/sdk/graph"
"git.novox.be/novox/mesh-sdk/go/views"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+1 -1
View File
@@ -31,7 +31,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-sdk/go/views"
"mesh-delivery-assigner/internal/assign"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+1 -1
View File
@@ -22,7 +22,7 @@ import (
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+1 -1
View File
@@ -34,7 +34,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-delivery/sdk/graph"
"mesh-delivery-assigner/internal/assign"
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go
+1 -1
View File
@@ -19,7 +19,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
)
// The assigner's tools (novox/hq ADR 0546), served through the tool runner from its bundle: a process never serves (ADR
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go
+2 -2
View File
@@ -18,9 +18,9 @@ import (
"git.novox.be/novox/mesh-sdk/go/views"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"git.novox.be/novox/mesh-delivery/sdk/graph"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
"mesh-delivery-assigner/internal/assign"
plc "mesh-delivery-assigner/internal/placement"
modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go
+1 -1
View File
@@ -21,7 +21,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
"mesh-delivery-builder/internal/versions"
)
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+1 -1
View File
@@ -19,7 +19,7 @@ import (
"syscall"
"time"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms"
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
"mesh-delivery-builder/internal/versions"
)
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
)
// EngineReadsState is the builder's state of what each counted node's engine states it reads (novox/hq ADR 0618 §2, ADR
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go
+2 -2
View File
@@ -14,7 +14,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats.go"
@@ -28,7 +28,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+1 -1
View File
@@ -18,7 +18,7 @@ import (
"syscall"
"time"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/holds.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+1 -1
View File
@@ -30,7 +30,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-indexer/internal/index"
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/reads.go
+1 -1
View File
@@ -10,7 +10,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"mesh-delivery-indexer/internal/index"
)
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go
+2 -2
View File
@@ -12,9 +12,9 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-sdk/go/process"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"git.novox.be/novox/mesh-delivery/sdk/graph"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
"git.novox.be/novox/mesh-sdk/go/views"
"mesh-delivery-indexer/internal/index"
modules/mesh-delivery-indexer/internal/index/holds.go
+2 -2
View File
@@ -30,7 +30,7 @@ import (
//
// "Counted" is ADR 0624 §1's set: every node the placement names, present or away, each read as its engine last stated
// through the node-engine seat's verb manifest-reads (a node with no statement read as the baseline; the indexer's state
// engine-reads, sdk/enginereads). Until the placement is read no node is counted and every hold stays. The hold moves
// engine-reads, nodeengineclient/enginereads). Until the placement is read no node is counted and every hold stays. The hold moves
// on the event manifest-reads-changed, the placement's change, a person's ask again, and a merge; never on a timer.
const (
// HoldsState is the indexer's state of held entries: the bucket mesh-delivery-indexer_holds, keyed as the modules state.
@@ -38,7 +38,7 @@ const (
HoldsBucket = Module + "_" + HoldsState
// HoldsTool reads the holds, asks one again now, or moves one by a person's word.
HoldsTool = "indexer_holds"
// EngineReadsState is the indexer's state of what each counted node's engine states it reads (sdk/enginereads).
// EngineReadsState is the indexer's state of what each counted node's engine states it reads (nodeengineclient/enginereads).
EngineReadsState = "engine-reads"
EngineReadsBucket = Module + "_" + EngineReadsState
// EngineReadsTool lists, asks again and drops those statements.
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+1 -1
View File
@@ -12,7 +12,7 @@ import (
"errors"
"fmt"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"io"
"log"
"os"
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
+1 -1
View File
@@ -19,7 +19,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-sdk/go/views"
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go
+1 -1
View File
@@ -16,7 +16,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-contracts/identity-and-access/go/clients/meshvaultclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-sdk/go/keys"
modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"git.novox.be/novox/mesh-delivery/sdk/trunk"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views"
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go
+1 -1
View File
@@ -18,7 +18,7 @@ import (
"syscall"
"time"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views"
modules/mesh-delivery-watchman/internal/watch/finding.go
+1 -1
View File
@@ -5,7 +5,7 @@ import (
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
)
// The watchman is an observer of the seat mesh-watchdogs (novox/hq ADR 0564 §2, §4): what it observes wrong it writes as a
modules/mesh-delivery-watchman/internal/watch/findings_test.go
+2 -2
View File
@@ -18,7 +18,7 @@ import (
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
)
func mustEncode(t *testing.T, f Finding) []byte {
@@ -244,7 +244,7 @@ func TestTheWatchmansFindings(t *testing.T) {
w.NotHeard("membership", "a consumer not there yet is not opened", errors.New("no bucket"))
w.Heard("EVENTS")
w.Heard("membership")
// sdk/heard's Writer, as the indexer, the assigner and the builder use it: Bind writes the finding with why
// meshconditionsclient/heard Writer, as the indexer, the assigner and the builder use it: Bind writes the finding with why
// when the consumer does not bind and answers the error, deletes it when it binds, and with no findings
// state says it in the log alone.
at := time.Date(2026, 11, 4, 17, 0, 0, 0, time.UTC)
sdk/enginereads/enginereads.go
-643
View File
@@ -1,643 +0,0 @@
// Package enginereads is how a mesh-delivery part learns what each node's engine reads of a manifest (novox/hq ADR 0618
// §3, §4): it asks the node seat node-engine's verb manifest-reads, on the part's start, on the seat's event
// manifest-reads-changed for that node and when a person asks again, and keeps each node's last answer in a state of its
// own, one key per node, so a node that is away keeps the statement it last made (ADR 0624 §1). The assigner, the
// builder and the delivery view share it; it is mesh-delivery's, not mesh-sdk's (ADR 0593 §1, ADR 0608 §5).
package enginereads
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
)
// The verb and the event as the seat's contract names them (mesh-contracts seats/nodeengine), as a manifest names them,
// and the subject the event is said on with the engine's machine last.
const (
Verb = nodeengine.VerbManifestReads
Event = nodeengine.EventManifestReadsChanged
Invoke = "seat:" + nodeengine.Seat + "." + Verb
Consumes = nodeengine.Seat + "." + Event
)
// SubjectPrefix is the event's subject up to the machine, its last token.
var SubjectPrefix = strings.TrimSuffix(nodeengine.SubjectManifestReadsChanged("*"), "*")
// The reasons an engine says its reads, in the event's why (ADR 0618 §4).
const (
// ReadsStarted: the engine started, or what it states changed.
ReadsStarted = "started"
// ReadsReturned: the engine's node returned after being away.
ReadsReturned = "returned"
)
// ContractVersion is the layout version of the verb's answer and the event's body this reader reads.
const ContractVersion = 1
// checkNode refuses a node that is no subject token.
func checkNode(node string) error {
if node == "" || strings.ContainsAny(node, ".*> \t\n") {
return fmt.Errorf("%q is no node", node)
}
return nil
}
// checkReads refuses an empty reader capability.
func checkReads(node string, reads []string) error {
for _, c := range reads {
if strings.TrimSpace(c) == "" {
return fmt.Errorf("the reads of %s name an empty reader capability", node)
}
}
return nil
}
// DecodeAnswer reads the verb's answer as the contract lays it out, and judges it: the layout's version, the node a
// name, no empty capability, and its stated-at an RFC 3339 time (the contract carries no checks; its reader does).
func DecodeAnswer(raw []byte) (nodeengine.ManifestReadsAnswer, time.Time, error) {
var a nodeengine.ManifestReadsAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return a, time.Time{}, err
}
if a.V != ContractVersion {
return a, time.Time{}, fmt.Errorf("the answer's layout is version %d, not %d", a.V, ContractVersion)
}
if err := checkNode(a.Node); err != nil {
return a, time.Time{}, err
}
if err := checkReads(a.Node, a.ManifestReads); err != nil {
return a, time.Time{}, err
}
at, err := time.Parse(time.RFC3339Nano, a.StatedAt)
if err != nil {
return a, time.Time{}, fmt.Errorf("stated-at %q is no RFC 3339 time", a.StatedAt)
}
return a, at, nil
}
// DecodeChanged reads the event's body as the contract lays it out, and judges it as DecodeAnswer does, its why one of
// ReadsStarted and ReadsReturned.
func DecodeChanged(raw []byte) (nodeengine.ManifestReadsChangedEvent, error) {
var e nodeengine.ManifestReadsChangedEvent
if err := json.Unmarshal(raw, &e); err != nil {
return e, err
}
if e.V != ContractVersion {
return e, fmt.Errorf("the event's layout is version %d, not %d", e.V, ContractVersion)
}
if err := checkNode(e.Node); err != nil {
return e, err
}
if e.Why != ReadsStarted && e.Why != ReadsReturned {
return e, fmt.Errorf("the reads of %s changed for %q, not %s or %s", e.Node, e.Why, ReadsStarted, ReadsReturned)
}
if err := checkReads(e.Node, e.ManifestReads); err != nil {
return e, err
}
if _, err := time.Parse(time.RFC3339Nano, e.At); err != nil {
return e, fmt.Errorf("at %q is no RFC 3339 time", e.At)
}
return e, nil
}
// Where a node's statement came from.
const (
// FromVerb: the node's engine answered the verb.
FromVerb = "verb"
// FromRecord: the engine did not answer it (an engine built before the verb, or nobody answered) and the node's record
// `<node>.node` was read once in its place (the fallback, until every counted engine serves the verb; issue 774).
FromRecord = "record"
// FromNone: neither said anything, so the node is read as stating the baseline (ADR 0614 §4).
FromNone = "none"
)
// Version is the version of a statement this package writes and the only one it reads.
const Version = 1
// Statement is one node's key: what its engine states it reads, when it stated it, when it was asked, where it came
// from, whether the engine answered that ask (it is present), and, while a person's ask again waits, that ask.
type Statement struct {
V int `json:"v"`
Node string `json:"node"`
ManifestReads []string `json:"manifest-reads,omitempty"`
StatedAt time.Time `json:"stated-at,omitzero"`
AskedAt time.Time `json:"asked-at,omitzero"`
From string `json:"from"`
Answered bool `json:"answered"`
Words string `json:"words,omitempty"`
AskAgain *Again `json:"ask-again,omitempty"`
}
// Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state.
type Again struct {
By string `json:"by"`
Why string `json:"why"`
At time.Time `json:"at"`
}
// Encode is s as JSON at Version.
func Encode(s Statement) ([]byte, error) {
if s.Node == "" || strings.ContainsAny(s.Node, ".*> ") {
return nil, fmt.Errorf("enginereads: %q is no node", s.Node)
}
s.V = Version
return json.Marshal(s)
}
// Decode reads a statement; it refuses another version and a value naming no node.
func Decode(raw []byte) (Statement, error) {
var s Statement
if err := json.Unmarshal(raw, &s); err != nil {
return Statement{}, err
}
if s.V != Version {
return Statement{}, fmt.Errorf("enginereads: version %d, not %d", s.V, Version)
}
if s.Node == "" {
return Statement{}, errors.New("enginereads: a statement names no node")
}
return s, nil
}
// Caller calls a seat's verb on one node: process.Conn.CallOn.
type Caller func(node, key string, body any) (json.RawMessage, error)
// Ask asks node's engine the verb once and says what it answered. An engine that answered the call with an error is
// present and states nothing through it; one that did not answer is not present.
func Ask(call Caller, node string, now time.Time) Statement {
s := Statement{Node: node, AskedAt: now.UTC(), From: FromNone}
raw, err := call(node, Invoke, map[string]any{})
var answered *busretry.Answered
switch {
case err == nil:
a, at, derr := DecodeAnswer(raw)
if derr == nil && a.Node != node {
derr = fmt.Errorf("it names the node %s", a.Node)
}
if derr != nil {
s.Answered, s.Words = true, "the engine answered what is no statement: "+derr.Error()
return s
}
s.ManifestReads, s.StatedAt, s.From, s.Answered = slices.Clone(a.ManifestReads), at.UTC(), FromVerb, true
case errors.As(err, &answered):
s.Answered, s.Words = true, "the engine does not serve "+Verb+": "+answered.Words
default:
s.Words = "nobody answered " + Verb + ": " + err.Error()
}
return s
}
// NodeOf is the node an event of manifest-reads-changed is about, read from its subject's last token and held equal to
// its body's; ok false for a subject of another event.
func NodeOf(subject string, data []byte) (string, bool, error) {
node, ok := strings.CutPrefix(subject, SubjectPrefix)
if !ok || node == "" || strings.Contains(node, ".") {
return "", false, nil
}
e, err := DecodeChanged(data)
if err != nil {
return node, true, err
}
if e.Node != node {
return node, true, fmt.Errorf("the event on %s names the node %s", subject, e.Node)
}
return node, true, nil
}
// Keeper keeps every node's statement for one part: in memory, and in the part's state when Put is set, so a statement
// outlives the part's restart. Nothing in it waits on a timer: a node is asked when the part says so.
type Keeper struct {
// Call asks the verb; nil asks nothing (every node then reads as the fallback or the baseline).
Call Caller
// Fallback reads a node's record `<node>.node` once where its engine did not answer the verb; nil reads none.
Fallback func(node string) ([]string, bool)
// Put writes a node's key of the part's state, Delete removes it; nil keeps the statements in memory only.
Put func(node string, value []byte) error
Delete func(node string) error
// Changed is told after a node's statement changed.
Changed func()
Logf func(string, ...any)
Now func() time.Time
// Sync asks in the caller's goroutine (tests); otherwise each ask runs on its own.
Sync bool
mu sync.Mutex
by map[string]Statement
asking map[string]bool
}
func (k *Keeper) logf(f string, a ...any) {
if k.Logf != nil {
k.Logf(f, a...)
}
}
func (k *Keeper) now() time.Time {
if k.Now != nil {
return k.Now()
}
return time.Now()
}
// AskNode asks node's engine, unless an ask of it runs already; why is said in the log.
func (k *Keeper) AskNode(node, why string) {
k.mu.Lock()
if k.asking == nil {
k.asking = map[string]bool{}
}
if k.asking[node] {
k.mu.Unlock()
return
}
k.asking[node] = true
k.mu.Unlock()
if k.Sync {
k.ask(node, why)
return
}
go k.ask(node, why)
}
func (k *Keeper) ask(node, why string) {
defer func() {
k.mu.Lock()
delete(k.asking, node)
k.mu.Unlock()
}()
s := Statement{Node: node, AskedAt: k.now().UTC(), From: FromNone, Words: "nothing asks the verb"}
if k.Call != nil {
s = Ask(k.Call, node, k.now())
}
if s.From != FromVerb && k.Fallback != nil {
if reads, ok := k.Fallback(node); ok {
s.ManifestReads, s.From = slices.Clone(reads), FromRecord
k.logf("%s on %s (%s): %s; its record %s.node is read once in its place (until every counted engine serves the verb)",
Verb, node, why, s.Words, node)
}
}
if s.From == FromNone {
k.logf("%s on %s (%s): %s; the node reads as its last statement, or the baseline", Verb, node, why, s.Words)
}
k.take(s)
}
// take keeps s, unless the node did not answer and an earlier statement stands: an away node keeps its last one (ADR 0624
// §1), marked as not answering.
func (k *Keeper) take(s Statement) {
k.mu.Lock()
if k.by == nil {
k.by = map[string]Statement{}
}
if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone {
old.Answered, old.AskedAt, old.Words, old.AskAgain = false, s.AskedAt, s.Words, nil
s = old
}
k.by[s.Node] = s
k.mu.Unlock()
if k.Put != nil {
raw, err := Encode(s)
if err == nil {
err = k.Put(s.Node, raw)
}
if err != nil {
k.logf("the statement of %s could not be kept: %v", s.Node, err)
}
}
if k.Changed != nil {
k.Changed()
}
}
// Heard takes a change of the part's own state in: at the start, every node's last statement; later, a person's ask again
// (asked now) or a key a person dropped (forgotten). The part's own writes come back here and change nothing.
func (k *Keeper) Heard(node string, raw []byte, deleted bool) {
if deleted {
k.mu.Lock()
_, had := k.by[node]
delete(k.by, node)
k.mu.Unlock()
if had && k.Changed != nil {
k.Changed()
}
return
}
s, err := Decode(raw)
if err != nil || s.Node != node {
k.logf("the key %s of the engine reads is no statement of it, left: %v", node, err)
return
}
if s.AskAgain != nil {
k.AskNode(node, fmt.Sprintf("asked again by %s: %s", s.AskAgain.By, s.AskAgain.Why))
return
}
k.mu.Lock()
if k.by == nil {
k.by = map[string]Statement{}
}
_, had := k.by[node]
if !had {
k.by[node] = s
}
k.mu.Unlock()
if !had && k.Changed != nil {
k.Changed()
}
}
// Forget drops a node the part no longer counts (it left the placement): its key too.
func (k *Keeper) Forget(node string) {
k.mu.Lock()
_, had := k.by[node]
delete(k.by, node)
k.mu.Unlock()
if had && k.Delete != nil {
if err := k.Delete(node); err != nil {
k.logf("the statement of %s could not be removed: %v", node, err)
}
}
}
// Known says whether the part holds a statement of node, from any source.
func (k *Keeper) Known(node string) bool {
k.mu.Lock()
defer k.mu.Unlock()
_, ok := k.by[node]
return ok
}
// Statements is every node's statement, by node.
func (k *Keeper) Statements() map[string]Statement {
k.mu.Lock()
defer k.mu.Unlock()
out := make(map[string]Statement, len(k.by))
for n, s := range k.by {
out[n] = s
}
return out
}
// Reads is, of the nodes given, what each states it reads: its last statement, nil (the baseline) for a node with none.
func (k *Keeper) Reads(nodes []string) map[string][]string {
k.mu.Lock()
defer k.mu.Unlock()
out := make(map[string][]string, len(nodes))
for _, n := range nodes {
out[n] = slices.Clone(k.by[n].ManifestReads)
}
return out
}
// Present is, of the nodes given, those whose engine answered when last asked, with what each states.
func (k *Keeper) Present(nodes []string) map[string][]string {
k.mu.Lock()
defer k.mu.Unlock()
out := map[string][]string{}
for _, n := range nodes {
if s, ok := k.by[n]; ok && s.Answered {
out[n] = slices.Clone(s.ManifestReads)
}
}
return out
}
// State is what the tool needs of the part's state: the runtime's (stdio.KeptState through KeptState), or a test's.
type State interface {
Keys() ([]string, error)
Get(key string) ([]byte, uint64, bool, error)
Put(key string, value []byte) (uint64, error)
Delete(key string) error
// Await waits at most bound for a revision of key after after whose value done says is the answer.
Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error)
}
// Read lists every node's statement in st, by node name, or one node's; a key that does not read is named with why.
func Read(st State, node string) (map[string]any, error) {
keys := []string{node}
if node == "" {
var err error
if keys, err = st.Keys(); err != nil {
return nil, err
}
}
sort.Strings(keys)
nodes := map[string]any{}
for _, key := range keys {
raw, _, found, err := st.Get(key)
switch {
case err != nil:
return nil, err
case !found:
nodes[key] = map[string]any{"from": FromNone, "words": "no statement is kept: read as the baseline"}
continue
}
s, derr := Decode(raw)
if derr != nil {
nodes[key] = map[string]any{"unread": derr.Error()}
continue
}
nodes[key] = s
}
return map[string]any{"nodes": nodes}, nil
}
// AskAgainOf writes node's key with a person's ask again and waits at most bound for the part's answer: the next
// statement it writes there.
func AskAgainOf(st State, node, by, why string, now time.Time, bound time.Duration) (Statement, error) {
if node == "" || strings.ContainsAny(node, ".*> ") {
return Statement{}, fmt.Errorf("%q is no node", node)
}
if strings.TrimSpace(why) == "" {
return Statement{}, errors.New("say why it is asked again")
}
s := Statement{Node: node, From: FromNone}
if raw, _, found, err := st.Get(node); err != nil {
return Statement{}, err
} else if found {
if old, err := Decode(raw); err == nil {
s = old
}
}
s.AskAgain = &Again{By: by, Why: why, At: now.UTC()}
raw, err := Encode(s)
if err != nil {
return Statement{}, err
}
rev, err := st.Put(node, raw)
if err != nil {
return Statement{}, err
}
answer, err := st.Await(node, rev, func(v []byte) bool {
got, err := Decode(v)
return err == nil && got.AskAgain == nil
}, bound)
if err != nil {
return Statement{}, fmt.Errorf("the part did not answer the ask again of %s within %s: %w", node, bound, err)
}
return Decode(answer)
}
// KeptState is a part's state as the runtime keeps it, for the tool.
func KeptState(s stdio.KeptState) State { return keptState{s} }
type keptState struct{ s stdio.KeptState }
func (k keptState) Keys() ([]string, error) { return k.s.Keys() }
func (k keptState) Get(key string) ([]byte, uint64, bool, error) {
e, err := k.s.Get(key)
if err != nil || e == nil {
return nil, 0, false, err
}
return e.Value, e.Revision, true, nil
}
func (k keptState) Put(key string, value []byte) (uint64, error) {
return k.s.Put(key, json.RawMessage(value))
}
func (k keptState) Delete(key string) error { return k.s.Delete(key) }
// Await watches key for the answer, and reads it once too in case the watch heard it before it began; a context's bound,
// no timer (ADR 0437 §8).
func (k keptState) Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error) {
heard := make(chan []byte, 1)
if err := k.s.Watch(key, func(c stdio.StateChange) error {
if c.Op == "put" && c.Revision > after && done(c.Value) {
select {
case heard <- c.Value:
default:
}
}
return nil
}); err != nil {
return nil, err
}
if e, err := k.s.Get(key); err == nil && e != nil && e.Revision > after && done(e.Value) {
return e.Value, nil
}
wait, cancel := context.WithTimeout(context.Background(), bound)
defer cancel()
select {
case v := <-heard:
return v, nil
case <-wait.Done():
return nil, errors.New("it is not running, or cannot read the verb's answer")
}
}
// AnswerWithin is how long the tool waits for the part to answer an ask again: the verb's own call bound and a margin.
var AnswerWithin = 25 * time.Second
// Tool is a part's tool over its state of engine reads (tools-first, ADR 0618 §3): read every node's statement or one,
// ask one node's engine again now, and, where drops, drop one node's key (a node the part keeps by mistake). owner names
// the part in the words.
func Tool(name, owner string, st State, drops bool, now func() time.Time) stdio.Tool {
props := map[string]any{
"node": map[string]any{"type": "string", "description": "one node; every node when left out"},
"ask_again": map[string]any{"type": "string", "description": "\"true\": ask the node's engine " + Verb + " again now, and answer what it states"},
"why": map[string]any{"type": "string", "description": "why it is asked again or dropped, in plain words"},
}
description := "What each node's engine states it reads of a manifest (manifest words, forms and seat names, novox/hq ADR 0618 §3), " +
"as " + owner + " last asked the node-engine seat's verb " + Verb + " and uses it: per node its capabilities, when the engine " +
"stated them and when it was asked, whether it answered (present) and where the statement came from: verb, record (the " +
"engine did not answer and its node record was read once in its place) or none (read as the baseline). A node that is " +
"away keeps its last statement. With node and ask_again, " + owner + " asks that node's engine again and answers its new " +
"statement."
if drops {
props["drop"] = map[string]any{"type": "string", "description": "\"true\": drop the node's key, so it reads as stating the baseline until its engine is asked again"}
description += " With node and drop, the node's key is dropped."
}
return stdio.Tool{Name: name, Description: description, Input: map[string]any{"type": "object", "properties": props},
Run: func(args map[string]any) (any, error) {
str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
node := str("node")
switch {
case str("ask_again") == "true":
return AskAgainOf(st, node, stdio.CallerOf(args), str("why"), now(), AnswerWithin)
case drops && str("drop") == "true":
if node == "" || str("why") == "" {
return nil, errors.New("drop takes node and why")
}
if err := st.Delete(node); err != nil {
return nil, err
}
return map[string]any{"node": node, "dropped": true,
"next": "the node reads as stating the baseline until its engine is asked again (ask_again, its event, or " + owner + "'s start)"}, nil
}
return Read(st, node)
}}
}
// How a node answered a survey (Survey).
const (
// Stated: the engine answered the verb.
Stated = "stated"
// NotServed: the engine answered, but does not serve the verb (an engine built before it).
NotServed = "not served"
// Away: nobody answered within the bound: the node is away, or its engine does not serve the verb and nothing else
// answers for it.
Away = "away"
)
// SurveyRow is one node's row of a survey: how it answered, its statement when it stated one, and its record's reads
// where it did not (the fallback the assigner reads in its place).
type SurveyRow struct {
Node string `json:"node"`
Answer string `json:"answer"`
ManifestReads []string `json:"manifest-reads,omitempty"`
StatedAt time.Time `json:"stated-at,omitzero"`
Record []string `json:"record,omitempty"`
RecordRead bool `json:"record-read,omitempty"`
Words string `json:"words,omitempty"`
}
// Survey asks every node's engine the verb at once, each ask bounded by bound, so the whole survey takes at most about
// bound however many nodes there are; a node that did not answer is named with its record's reads where record reads
// one. Sorted by node.
func Survey(call Caller, nodes []string, record func(node string) ([]string, bool), bound time.Duration, now time.Time) []SurveyRow {
rows := make([]SurveyRow, len(nodes))
var wg sync.WaitGroup
for i, node := range nodes {
wg.Add(1)
go func() {
defer wg.Done()
got := make(chan Statement, 1)
go func() { got <- Ask(call, node, now) }()
var s Statement
wait, cancel := context.WithTimeout(context.Background(), bound)
defer cancel()
select {
case s = <-got:
case <-wait.Done():
s = Statement{Node: node, Words: fmt.Sprintf("no answer within %s", bound)}
}
row := SurveyRow{Node: node, Words: s.Words}
switch {
case s.From == FromVerb:
row.Answer, row.ManifestReads, row.StatedAt = Stated, s.ManifestReads, s.StatedAt
case s.Answered:
row.Answer = NotServed
default:
row.Answer = Away
}
if row.Answer != Stated && record != nil {
row.Record, row.RecordRead = record(node)
}
rows[i] = row
}()
}
wg.Wait()
sort.Slice(rows, func(i, j int) bool { return rows[i].Node < rows[j].Node })
return rows
}
sdk/findings/findings.go
-88
View File
@@ -1,88 +0,0 @@
// Package findings is a part's own bucket `findings` as the state the seat mesh-conditions' client writes into
// (meshconditionsclient.FindingState), so a part gives mesh-sdk's views the FindingWriter of the seat's layout without
// each writing the same adapter over its bucket (novox/hq ADR 0564 §2, ADR 0635 §4).
package findings
import (
"context"
"encoding/json"
"errors"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-sdk/go/views"
)
// ErrUnbound says the bucket is not on the bus yet, so the finding is not written (the part says it in its log).
var ErrUnbound = errors.New("the findings state is not bound yet")
// State is the bucket the part's state `findings` is, given when it is bound (process.AwaitedBucket.KV): nil is not bound yet.
type State struct {
Ctx context.Context
KV func() jetstream.KeyValue
}
var _ meshconditionsclient.FindingState = State{}
// Put writes one value under its key as JSON.
func (s State) Put(key string, value any) (uint64, error) {
kv := s.KV()
if kv == nil {
return 0, ErrUnbound
}
data, err := json.Marshal(value)
if err != nil {
return 0, err
}
return kv.Put(s.Ctx, key, data)
}
// Delete clears one key.
func (s State) Delete(key string) error {
kv := s.KV()
if kv == nil {
return ErrUnbound
}
return kv.Delete(s.Ctx, key)
}
// Writer is the part's views.FindingWriter over its bucket; the function is what views.ProcessReader.Findings and the
// emitter's Findings take.
func Writer(ctx context.Context, kv func() jetstream.KeyValue) views.FindingWriter {
return meshconditionsclient.FindingWriter{State: State{Ctx: ctx, KV: kv}}
}
// Funcs is a findings state held as the two functions a part keeps over it (the assigner's findings, which also say
// its conditions): a value is put as JSON.
type Funcs struct {
PutBytes func(key string, value []byte) error
DeleteKey func(key string) error
}
var _ meshconditionsclient.FindingState = Funcs{}
// Put writes one value under its key as JSON; the state not bound (no function) is ErrUnbound.
func (f Funcs) Put(key string, value any) (uint64, error) {
if f.PutBytes == nil {
return 0, ErrUnbound
}
data, err := json.Marshal(value)
if err != nil {
return 0, err
}
return 0, f.PutBytes(key, data)
}
// Delete clears one key.
func (f Funcs) Delete(key string) error {
if f.DeleteKey == nil {
return ErrUnbound
}
return f.DeleteKey(key)
}
// WriterOver is the part's views.FindingWriter over its two functions.
func WriterOver(put func(string, []byte) error, del func(string) error) views.FindingWriter {
return meshconditionsclient.FindingWriter{State: Funcs{PutBytes: put, DeleteKey: del}}
}
sdk/findings/findings_test.go
-71
View File
@@ -1,71 +0,0 @@
package findings
import (
"context"
"encoding/json"
"errors"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/views"
)
// fakeKV is the two calls a findings state makes of its bucket.
type fakeKV struct {
jetstream.KeyValue
held map[string][]byte
}
func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
f.held[key] = value
return uint64(len(f.held)), nil
}
func (f *fakeKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
delete(f.held, key)
return nil
}
// A part's finding is written in the seat mesh-conditions' layout under the key the condition keeper reads, cleared by the
// same finding, and refused while the bucket is not bound (the part says that in its log).
func TestAFindingIsWrittenInTheSeatsLayoutAndClearedAndRefusedUnbound(t *testing.T) {
kv := &fakeKV{held: map[string][]byte{}}
var bound jetstream.KeyValue
w := Writer(context.Background(), func() jetstream.KeyValue { return bound })
f := views.Finding{Observer: "mesh-delivery-teller", Kind: views.UnansweredKind, ID: "mesh-delivery-teller.asks", Summary: "s", Headline: "h",
Explanation: "e", At: time.Date(2026, 11, 5, 15, 0, 0, 0, time.UTC)}
if err := w.Put(f); !errors.Is(err, ErrUnbound) {
t.Fatalf("an unbound bucket took the finding: %v", err)
}
bound = kv
if err := w.Put(f); err != nil {
t.Fatal(err)
}
var got struct{ Key, Kind, Observer, Source string }
for key, value := range kv.held {
if err := json.Unmarshal(value, &got); err != nil || got.Key != key || got.Kind != views.UnansweredKind ||
got.Observer != f.Observer || got.Source != f.Observer+"."+f.Kind {
t.Fatalf("%s holds %s (%v)", key, value, err)
}
}
if len(kv.held) != 1 {
t.Fatalf("the bucket holds %d findings, want 1", len(kv.held))
}
if err := w.Delete(f); err != nil || len(kv.held) != 0 {
t.Fatalf("the finding is not cleared: %v, %d left", err, len(kv.held))
}
// The same finding over two functions (a part whose findings also say its conditions).
held := map[string][]byte{}
o := WriterOver(func(k string, v []byte) error { held[k] = v; return nil }, func(k string) error { delete(held, k); return nil })
if err := o.Put(f); err != nil || len(held) != 1 {
t.Fatalf("over functions: %v, %d held", err, len(held))
}
if err := o.Delete(f); err != nil || len(held) != 0 {
t.Fatalf("over functions not cleared: %v, %d left", err, len(held))
}
if err := WriterOver(nil, nil).Put(f); !errors.Is(err, ErrUnbound) {
t.Fatalf("no functions took the finding: %v", err)
}
}
sdk/go.mod
-3
View File
@@ -4,8 +4,6 @@ go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
require git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000
require (
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
@@ -19,6 +17,5 @@ require (
)
require (
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 // indirect
)
sdk/heard/heard.go
-115
View File
@@ -1,115 +0,0 @@
// Package heard is the finding a part of mesh-delivery writes while one of its consumers or watches does not hear the
// family it follows (novox/hq ADR 0615 §5; review round 1 of the registry milestone): the consumer not there yet, a
// hearing that could not start, a watch that cannot be opened. Without it the part's health stays healthy and its
// status says answered while the events it waits on are lost to it. The finding names the observer and the family,
// says what is not heard and why, and is deleted when the consumer binds: at the part's next start, or where the part
// binds again at a change of its membership, then; the condition keeper raises and clears the condition from it (ADR 0564 §2).
package heard
import (
"encoding/json"
"fmt"
"strings"
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-sdk/go/views"
)
// Kind is the finding's kind.
const Kind = "family-not-heard"
// ID is the finding's subject: `<observer>.<family>`, the family by the name the part follows it under (a stream, an
// event, a state's watch).
func ID(observer, family string) string { return observer + "." + family }
// Key is the finding's key in the observer's findings state.
func Key(observer, family string) string {
return meshconditionsclient.Key(meshconditionsclient.ScopeModule, ID(observer, family), Kind)
}
// Words are the finding's words: its summary, headline and explanation. consequence says what the observer misses while
// it does not hear family ("the placement is the first view's alone").
func Words(observer, family, consequence string, why error) (summary, headline, explanation string) {
reason := "it gave no reason"
if why != nil {
reason = strings.TrimSpace(why.Error())
}
summary = fmt.Sprintf("%s does not hear %s: %s (%s)", observer, family, consequence, reason)
headline = fmt.Sprintf("%s does not hear %s", observer, family)
explanation = fmt.Sprintf("%s follows %s on a consumer of its own, and that consumer is not there or could not be heard, so %s. "+
"This clears when the consumer binds: at %s's next start, or sooner where %s binds again at a change of its membership. "+
"Look at whether the grants part made %s's consumers, then start %s again if it does not bind by itself.", observer, family, consequence, observer, observer, observer, observer)
return summary, headline, explanation
}
// Finding is observer's finding that it does not hear family, in the layout of the seat mesh-conditions' state findings.
func Finding(observer, family, consequence string, why error, at time.Time) (string, []byte, error) {
summary, headline, explanation := Words(observer, family, consequence, why)
v := meshconditionsclient.FindingFor(views.Finding{Observer: observer, Kind: Kind, ID: ID(observer, family), Summary: summary,
Headline: headline, Explanation: explanation, At: at})
if err := meshconditionsclient.CheckFinding(v); err != nil {
return v.Key, nil, err
}
value, err := json.Marshal(v)
return v.Key, value, err
}
// Writer writes and deletes the finding in the observer's findings state. Put or Delete nil writes nothing, said in the
// log, so the part still says it.
type Writer struct {
Observer string
Put func(key string, value []byte) error
Delete func(key string) error
Logf func(string, ...any)
Now func() time.Time
}
// Bind binds the observer's consumer of family (bind makes it and starts hearing it): the finding is deleted when it
// binds, and written, with why, when it does not. It answers bind's error, so a caller that cannot go on without the
// family still stops.
func (w Writer) Bind(family, consequence string, bind func() error) error {
if err := bind(); err != nil {
w.NotHeard(family, consequence, err)
return err
}
w.Heard(family)
return nil
}
// NotHeard says in the log and in the finding that the observer does not hear family.
func (w Writer) NotHeard(family, consequence string, why error) {
now := time.Now
if w.Now != nil {
now = w.Now
}
summary, _, _ := Words(w.Observer, family, consequence, why)
w.logf("%s", summary)
if w.Put == nil {
w.logf("the finding %s is not written: no findings state", Key(w.Observer, family))
return
}
key, value, err := Finding(w.Observer, family, consequence, why, now())
if err == nil {
err = w.Put(key, value)
}
if err != nil {
w.logf("the finding %s is not written: %v", key, err)
}
}
// Heard deletes the finding: the consumer of family is bound.
func (w Writer) Heard(family string) {
if w.Delete == nil {
return
}
if err := w.Delete(Key(w.Observer, family)); err != nil {
w.logf("the finding %s is not deleted: %v", Key(w.Observer, family), err)
}
}
func (w Writer) logf(format string, args ...any) {
if w.Logf != nil {
w.Logf(format, args...)
}
}