The router takes a module's condition on the operator channel's one subject (design 47 R3b: issues 1217, 1218) #369
@@ -450,6 +450,15 @@ const ConditionOwner = "mesh-controller"
|
||||
// grants that module alone. An error is returned only when the ask could not be recorded, so the queue
|
||||
// offers it again; a refusal is said to the asker and the work is taken.
|
||||
func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error {
|
||||
if verb == ConditionVerb {
|
||||
// Before the work lock: Condition takes it, and decides as for the controller's conditions.
|
||||
if !asks.UsableID(strings.ReplaceAll(asker, "-", "_")) {
|
||||
h.Logf("[messenger] a condition from %q was taken and dropped: not a module's name", asker)
|
||||
return nil
|
||||
}
|
||||
h.takeCondition(asker, body)
|
||||
return nil
|
||||
}
|
||||
h.warmRootFree(true) // fresh, before the work: an ask refused for no channel stays refused with its asker
|
||||
h.work.Lock()
|
||||
defer h.work.Unlock()
|
||||
|
||||
@@ -34,6 +34,8 @@ const bundleVar = "MESH_MESSENGER_TEST_BUNDLE"
|
||||
func TestMain(m *testing.M) {
|
||||
if os.Getenv(bundleVar) == "1" {
|
||||
kindsEvery = 200 * time.Millisecond
|
||||
StartGrace = 500 * time.Millisecond
|
||||
BurstWindow = 200 * time.Millisecond
|
||||
main()
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
@@ -38,6 +38,8 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
// The events, by their local names under the controller's seat.
|
||||
@@ -57,19 +59,22 @@ const (
|
||||
|
||||
// Condition is what this holder needs of one.
|
||||
type Condition struct {
|
||||
Key string
|
||||
Kind string
|
||||
Scope string
|
||||
ID string
|
||||
Machine string
|
||||
Severity string
|
||||
Summary string
|
||||
Headline string
|
||||
Explanation string
|
||||
Resolved string
|
||||
Needs string
|
||||
Actions []Action
|
||||
Source string
|
||||
Key string
|
||||
Kind string
|
||||
Scope string
|
||||
ID string
|
||||
Machine string
|
||||
Severity string
|
||||
Summary string
|
||||
Headline string
|
||||
Explanation string
|
||||
Resolved string
|
||||
Needs string
|
||||
Actions []Action
|
||||
Source string
|
||||
// Publisher is the module that said the condition on the operator channel's one subject; empty for the
|
||||
// controller's own, which a reading of the controller's conditions may end.
|
||||
Publisher string
|
||||
Resolver string
|
||||
Raised time.Time
|
||||
LastObserved time.Time
|
||||
@@ -341,3 +346,33 @@ func DecodeOpen(raw json.RawMessage) ([]Condition, error) {
|
||||
}
|
||||
return nil, fmt.Errorf("the conditions answer lists no conditions")
|
||||
}
|
||||
|
||||
// ConditionVerb is the verb a module says a condition with: `mesh.seat.operator-channel.accept.condition.<publisher>`
|
||||
// (novox/hq ADR 0437 §9).
|
||||
const ConditionVerb = "condition"
|
||||
|
||||
// takeCondition takes one condition a module said on the seat's one subject. It is decided exactly as the
|
||||
// controller's conditions are, by Condition: there is no second path. The publisher is the subject's last
|
||||
// token, which the bus grants that module alone, so it is not compared with the condition's key. A body that is
|
||||
// not a condition in the SDK's layout is logged by publisher, taken and dropped.
|
||||
func (h *Holder) takeCondition(publisher string, body []byte) {
|
||||
sc, err := asks.DecodeCondition(body)
|
||||
if err != nil {
|
||||
h.Logf("[messenger] a condition from %s was taken and dropped: %v", publisher, err)
|
||||
return
|
||||
}
|
||||
c := Condition{Key: sc.Key, Scope: sc.Subject.Scope, ID: sc.Subject.ID, Machine: sc.Subject.Machine,
|
||||
Severity: sc.Severity, Summary: sc.Summary, Headline: sc.Headline, Explanation: sc.Explanation,
|
||||
Resolved: sc.Resolved, Needs: sc.Needs, Source: publisher, Publisher: publisher}
|
||||
if i := strings.LastIndex(sc.Key, "."); i >= 0 {
|
||||
c.Kind = sc.Key[i+1:]
|
||||
}
|
||||
for _, a := range sc.Actions {
|
||||
c.Actions = append(c.Actions, Action{Label: a.Label, Verb: a.Verb, Machine: a.Machine, Arguments: a.Arguments})
|
||||
}
|
||||
if sc.State == asks.StateCleared {
|
||||
h.Condition(EventCleared, c)
|
||||
return
|
||||
}
|
||||
h.Condition(EventRaised, c)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// A condition said on the operator channel's one subject over the real bus (novox/hq ADR 0437 §9, issue 1218),
|
||||
// in the bed of bus_test.go. The fake runtime enforces no grant: that a module may say a condition only under its
|
||||
// own name is the broker's, proven in the controller (issue 1216).
|
||||
|
||||
// conditionBed starts the router with the desk able to send, hands it the conditions said on the one subject as
|
||||
// work, and keeps what the desk is sent and whether each work item was acknowledged.
|
||||
type conditionBed struct {
|
||||
r *busRuntime
|
||||
mu sync.Mutex
|
||||
desk []asks.Message
|
||||
acks []string // the answer of the router to each work item: "" when acknowledged
|
||||
}
|
||||
|
||||
func newConditionBed(t *testing.T) *conditionBed {
|
||||
t.Helper()
|
||||
b := &conditionBed{r: newBusRuntime(t)}
|
||||
b.r.setKinds(map[string][]string{"desktop": {"choice", "edit"}})
|
||||
if _, err := b.r.nc.Subscribe("mesh.seat."+asks.ChannelSeat+".accept.*.desktop", func(m *nats.Msg) {
|
||||
var msg asks.Message
|
||||
if json.Unmarshal(m.Data, &msg) == nil {
|
||||
b.mu.Lock()
|
||||
b.desk = append(b.desk, msg)
|
||||
b.mu.Unlock()
|
||||
}
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.condition.*", func(m *nats.Msg) {
|
||||
answer, err := b.r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}})
|
||||
var a struct {
|
||||
Error *struct {
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
}
|
||||
_ = json.Unmarshal(answer, &a)
|
||||
res := ""
|
||||
if err != nil {
|
||||
res = err.Error()
|
||||
} else if a.Error != nil {
|
||||
res = a.Error.Message
|
||||
}
|
||||
b.mu.Lock()
|
||||
b.acks = append(b.acks, res)
|
||||
b.mu.Unlock()
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.forwardEvents(t)
|
||||
b.r.start(t.TempDir())
|
||||
b.r.until("the bundle takes work and read the kinds", 20*time.Second, func() bool {
|
||||
b.r.mu.Lock()
|
||||
defer b.r.mu.Unlock()
|
||||
return b.r.taking && b.r.kindsRead > 0
|
||||
})
|
||||
b.r.until("the bundle subscribed to the channels' word", 20*time.Second, func() bool {
|
||||
b.r.mu.Lock()
|
||||
defer b.r.mu.Unlock()
|
||||
return b.r.subscribed
|
||||
})
|
||||
standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()})
|
||||
if err := b.r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "desktop"), standing); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
return b
|
||||
}
|
||||
|
||||
// forwardEvents hands the bundle the channels' word, as bus_test.go's forward does for it.
|
||||
func (b *conditionBed) forwardEvents(t *testing.T) {
|
||||
t.Helper()
|
||||
if _, err := b.r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) {
|
||||
key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1)
|
||||
_, _ = b.r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}})
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (b *conditionBed) say(t *testing.T, publisher string, c asks.Condition) {
|
||||
t.Helper()
|
||||
body, err := asks.EncodeCondition(c)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
subject, err := asks.ConditionSubject(publisher)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := b.r.nc.Publish(subject, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (b *conditionBed) record(key string) *Record {
|
||||
e, err := b.r.bucket("open").Get(context.Background(), kvKey(key))
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
var rec Record
|
||||
if json.Unmarshal(e.Value(), &rec) != nil {
|
||||
return nil
|
||||
}
|
||||
return &rec
|
||||
}
|
||||
|
||||
func (b *conditionBed) told() []asks.Message {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return append([]asks.Message(nil), b.desk...)
|
||||
}
|
||||
|
||||
func (b *conditionBed) acknowledged(n int) bool {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if len(b.acks) != n {
|
||||
return false
|
||||
}
|
||||
for _, a := range b.acks {
|
||||
if a != "" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func busCondition(key, id, headline, needs string) asks.Condition {
|
||||
verdict := asks.NothingToDo
|
||||
if needs != "" {
|
||||
verdict = asks.NeedsYou + needs
|
||||
}
|
||||
return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id},
|
||||
State: asks.StateRaised, Summary: "the merge waits on a check", Headline: headline, Needs: needs,
|
||||
Explanation: verdict + " It waits on a check."}
|
||||
}
|
||||
|
||||
// 1218/AC1 and 1218/AC2: a raised condition said on the one subject reaches the desk, is recorded and its work item
|
||||
// is acknowledged; its cleared form ends it. The bundle starts with a short grace (bus_test.go).
|
||||
func TestAConditionSaidOnTheOneSubjectReachesTheDeskAndIsClearedOverTheBus(t *testing.T) {
|
||||
b := newConditionBed(t)
|
||||
b.say(t, "mesh-delivery-teller", busCondition("merge.d-7.stalled", "d-7", "A merge waits on a check", "release it, or stop it."))
|
||||
// 1218/AC1: recorded, acknowledged, and told to the desk.
|
||||
b.r.until("the condition is recorded open", 30*time.Second, func() bool {
|
||||
rec := b.record("merge.d-7.stalled")
|
||||
return rec != nil && rec.Origin == "condition" && rec.Cleared.IsZero()
|
||||
})
|
||||
b.r.until("the work item is acknowledged", 30*time.Second, func() bool { return b.acknowledged(1) })
|
||||
b.r.until("the desk is sent the condition", 15*time.Second, func() bool {
|
||||
for _, m := range b.told() {
|
||||
if strings.Contains(m.Title+m.Body, "A merge waits on a check") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
time.Sleep(3 * time.Second)
|
||||
if n := b.nAcks(); n != 1 {
|
||||
t.Errorf("the work item was offered %d times, want once: it was not acknowledged", n)
|
||||
}
|
||||
// 1218/AC2: the cleared form, on the same subject, ends it.
|
||||
b.say(t, "mesh-delivery-teller", asks.Condition{Key: "merge.d-7.stalled", Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateCleared})
|
||||
b.r.until("the condition is recorded cleared", 30*time.Second, func() bool {
|
||||
rec := b.record("merge.d-7.stalled")
|
||||
return rec == nil || !rec.Cleared.IsZero()
|
||||
})
|
||||
}
|
||||
|
||||
func (b *conditionBed) nAcks() int {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return len(b.acks)
|
||||
}
|
||||
|
||||
// 1218/AC3: a warning whose needs is empty is recorded and reaches no channel. A second warning that needs the
|
||||
// operator, said after it, is told: when it is, the quiet one's turn has passed and it was not sent.
|
||||
func TestAWarningThatNeedsNothingReachesNoChannelOverTheBus(t *testing.T) {
|
||||
b := newConditionBed(t)
|
||||
b.say(t, "mesh-delivery-teller", busCondition("merge.d-8.quiet", "d-8", "A quiet merge note", ""))
|
||||
b.r.until("the quiet warning is recorded", 30*time.Second, func() bool { return b.record("merge.d-8.quiet") != nil })
|
||||
b.say(t, "mesh-delivery-teller", busCondition("merge.d-9.loud", "d-9", "A loud merge note", "release it, or stop it."))
|
||||
b.r.until("the warning that needs the operator is sent to the desk", 15*time.Second, func() bool {
|
||||
for _, m := range b.told() {
|
||||
if strings.Contains(m.Title+m.Body, "A loud merge note") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
for _, m := range b.told() {
|
||||
if strings.Contains(m.Title+m.Body, "A quiet merge note") {
|
||||
t.Errorf("a warning that needs nothing was sent: %+v", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
// sdkCondition is a raised warning in mesh-sdk's layout, as a module says it on the operator channel's one subject.
|
||||
func sdkCondition(needs string) asks.Condition {
|
||||
verdict := asks.NothingToDo
|
||||
if needs != "" {
|
||||
verdict = asks.NeedsYou + needs
|
||||
}
|
||||
return asks.Condition{Key: "merge.d-7.stalled", Severity: asks.SeverityWarning,
|
||||
Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateRaised, Summary: "the merge waits on a check",
|
||||
Headline: "A merge waits on a check", Needs: needs, Explanation: verdict + " It waits on a check."}
|
||||
}
|
||||
|
||||
func conditionBody(t *testing.T, c asks.Condition) json.RawMessage {
|
||||
t.Helper()
|
||||
b, err := asks.EncodeCondition(c)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func TestARaisedConditionSaidOnTheOneSubjectIsToldAsTheControllersIs(t *testing.T) { // 1217/AC1
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, sdkCondition("look at it."))); err != nil {
|
||||
t.Fatalf("not acknowledged: %v", err) // 1217/AC1: an error would offer the work again
|
||||
}
|
||||
open := h.Open()
|
||||
if len(open) != 1 || open[0].Key != "merge.d-7.stalled" || open[0].Subject != "merge d-7" || open[0].Origin != "condition" {
|
||||
t.Fatalf("open: %+v", open)
|
||||
}
|
||||
settle(h, c)
|
||||
if len(dt.sends) != 1 || len(tg.sends) != 0 {
|
||||
t.Fatalf("a warning goes to the desk when a session answers: desktop %d telegram %d", len(dt.sends), len(tg.sends))
|
||||
}
|
||||
dt.fail = errors.New("the account is not logged in")
|
||||
c.pass(DesktopWarningEvery)
|
||||
k := sdkCondition("look at it.")
|
||||
k.Key, k.Subject.ID = "merge.d-8.stalled", "d-8"
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, k)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
settle(h, c) // the desktop fails: handed to Telegram
|
||||
h.Flush()
|
||||
if len(tg.sends) != 1 {
|
||||
t.Fatalf("no session: telegram %d", len(tg.sends))
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheSameConditionTakenTwiceIsSaidOnce(t *testing.T) { // 1217/AC2
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
body := conditionBody(t, sdkCondition("look at it."))
|
||||
for range 2 {
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
settle(h, c)
|
||||
if len(dt.sends)+len(tg.sends) != 1 {
|
||||
t.Fatalf("said %d times, want once", len(dt.sends)+len(tg.sends))
|
||||
}
|
||||
if open := h.Open(); len(open) != 1 {
|
||||
t.Fatalf("open: %+v", open)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAClearedConditionIsClearedAsTheControllersIs(t *testing.T) { // 1217/AC3
|
||||
h, _, _, c, _ := newHolder(t)
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, sdkCondition("look at it."))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
settle(h, c)
|
||||
cleared := asks.Condition{Key: "merge.d-7.stalled", Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateCleared}
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, cleared)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if open := h.Open(); len(open) != 0 {
|
||||
t.Fatalf("still open: %+v", open)
|
||||
}
|
||||
// The same as the controller's clearing of the same key.
|
||||
h2, _, _, c2, _ := newHolder(t)
|
||||
h2.Condition(EventRaised, cond("merge.d-7.stalled", Warning, "the merge waits on a check"))
|
||||
settle(h2, c2)
|
||||
h2.Condition(EventCleared, Condition{Key: "merge.d-7.stalled"})
|
||||
if len(h2.Open()) != 0 {
|
||||
t.Fatal("the controller's clearing leaves it open")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAWarningThatNeedsNothingIsKeptQuiet(t *testing.T) { // 1217/AC4
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, sdkCondition(""))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
settle(h, c)
|
||||
if len(dt.sends)+len(tg.sends) != 0 {
|
||||
t.Fatalf("a quiet warning was told: desktop %d telegram %d", len(dt.sends), len(tg.sends))
|
||||
}
|
||||
if len(h.Open()) != 1 {
|
||||
t.Fatalf("it is still recorded: %+v", h.Open())
|
||||
}
|
||||
}
|
||||
|
||||
func TestABodyThatIsNotAConditionIsDroppedAndTheLogNamesThePublisher(t *testing.T) { // 1217/AC5
|
||||
for name, body := range map[string]string{
|
||||
"bad JSON": `{"v":1,`,
|
||||
"unknown version": `{"v":9,"key":"merge.d-7.stalled","subject":{"scope":"merge","id":"d-7"},"state":"cleared"}`,
|
||||
"no version": `{"key":"merge.d-7.stalled","subject":{"scope":"merge","id":"d-7"},"state":"cleared"}`,
|
||||
} {
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
var logged []string
|
||||
h.Logf = func(f string, a ...any) { logged = append(logged, strings.TrimSpace(fmt.Sprintf(f, a...))) }
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", json.RawMessage(body)); err != nil {
|
||||
t.Fatalf("%s: the work is taken and dropped, got %v", name, err)
|
||||
}
|
||||
settle(h, c)
|
||||
if len(h.Open()) != 0 || len(dt.sends)+len(tg.sends) != 0 {
|
||||
t.Fatalf("%s: something was recorded or told", name)
|
||||
}
|
||||
if !strings.Contains(strings.Join(logged, "\n"), "mesh-delivery-teller") {
|
||||
t.Fatalf("%s: the log does not name the publisher: %q", name, logged)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAModulesConditionStaysOpenWhenTheControllersConditionsAreRead(t *testing.T) { // 1217/AC7
|
||||
h, tg, dt, c, _ := newHolder(t)
|
||||
body := conditionBody(t, sdkCondition("look at it."))
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
settle(h, c)
|
||||
c.pass(time.Second)
|
||||
res := h.Sync(nil, c.now()) // the controller lists nothing: the teller's condition is not its to end
|
||||
if len(res.Gone) != 0 || len(h.Open()) != 1 {
|
||||
t.Fatalf("the reading ended a module's condition: gone %v, open %+v", res.Gone, h.Open())
|
||||
}
|
||||
if err := h.TakeAsk("condition", "mesh-delivery-teller", body); err != nil { // raised again: the same message
|
||||
t.Fatal(err)
|
||||
}
|
||||
settle(h, c)
|
||||
if len(dt.sends)+len(tg.sends) != 1 {
|
||||
t.Fatalf("said %d times, want once", len(dt.sends)+len(tg.sends))
|
||||
}
|
||||
}
|
||||
@@ -55,6 +55,14 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// BurstWindow: what arrives for a channel within it goes out as one message. A variable, so the test bundle
|
||||
// shortens it.
|
||||
var BurstWindow = 30 * time.Second
|
||||
|
||||
// StartGrace: after start, nothing but the urgent conditions open in the controller's state. A variable, as
|
||||
// kindsEvery is, so the test bundle shortens it.
|
||||
var StartGrace = time.Minute
|
||||
|
||||
const (
|
||||
RemindUrgent = time.Hour
|
||||
RemindWarning = 12 * time.Hour
|
||||
@@ -71,12 +79,8 @@ const (
|
||||
|
||||
// FreshFor: an event older than this by its own time is history — state, never a message.
|
||||
FreshFor = 10 * time.Minute
|
||||
// BurstWindow: what arrives for a channel within it goes out as one message.
|
||||
BurstWindow = 30 * time.Second
|
||||
// DesktopWarningEvery: warnings reach the desktop at most this often.
|
||||
DesktopWarningEvery = 15 * time.Minute
|
||||
// StartGrace: after start, nothing but the urgent conditions open in the controller's state.
|
||||
StartGrace = time.Minute
|
||||
// ResyncEvery: how often the open set is read from the controller again.
|
||||
ResyncEvery = 10 * time.Minute
|
||||
// RetryEvery: a channel that failed is tried again after this.
|
||||
@@ -184,8 +188,10 @@ type Record struct {
|
||||
Answered string `json:"answered,omitempty"`
|
||||
// Ask is the open ask about this record, `<asker>.<id>`: its options are this record's answers
|
||||
// (novox/hq ADR 0259).
|
||||
Ask string `json:"ask,omitempty"`
|
||||
Origin string `json:"origin"`
|
||||
Ask string `json:"ask,omitempty"`
|
||||
Origin string `json:"origin"`
|
||||
// Publisher is the module that said it on the operator channel's one subject; empty for the controller's.
|
||||
Publisher string `json:"publisher,omitempty"`
|
||||
More string `json:"more"`
|
||||
Raised time.Time `json:"raised"`
|
||||
SilencedTill time.Time `json:"silenced_till,omitempty"`
|
||||
@@ -228,8 +234,8 @@ func recordOf(c Condition) Record {
|
||||
return Record{Key: c.Key, Kind: c.Kind, Subject: c.SubjectWords(), Scope: c.Scope, SubjectID: c.ID, Machine: c.Machine,
|
||||
Severity: c.Severity, Summary: c.Summary,
|
||||
Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved, Needs: c.Needs, Actions: c.Actions,
|
||||
Origin: "condition",
|
||||
More: "conditions show " + c.Key, Raised: c.Raised, SilencedTill: c.SilencedTill, SilencedWhy: c.SilencedWhy}
|
||||
Origin: "condition", Publisher: c.Publisher,
|
||||
More: "conditions show " + c.Key, Raised: c.Raised, SilencedTill: c.SilencedTill, SilencedWhy: c.SilencedWhy}
|
||||
}
|
||||
|
||||
// takeWords keeps the plain words current, as the summary is: a condition's words follow what is seen.
|
||||
|
||||
@@ -60,12 +60,12 @@ func readManifest(t *testing.T) (manifest, string) {
|
||||
return m, string(raw)
|
||||
}
|
||||
|
||||
func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) {
|
||||
func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) { // 1217/AC6: condition beside ask and cancel
|
||||
m, _ := readManifest(t)
|
||||
verbs := []string{"open", "history", "notify", "tell", "asks"}
|
||||
want := []seat{
|
||||
{Name: asks.Seat, Scope: "mesh", Serves: verbs, Accepts: []string{"ask", "cancel"}, Emits: []string{"decided"},
|
||||
ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"asks"}},
|
||||
{Name: asks.Seat, Scope: "mesh", Serves: verbs, Accepts: []string{"ask", "cancel", "condition"}, Emits: []string{"decided"},
|
||||
ByCaller: []string{"ask", "cancel", "condition", "decided"}, Records: []string{"asks"}},
|
||||
{Name: asks.ChannelSeat, Scope: "mesh", Kinded: true, Accepts: []string{asks.Show, asks.Edit, asks.Send}},
|
||||
{Name: asks.IntakeSeat, Scope: "mesh", Kinded: true,
|
||||
Emits: []string{asks.Chosen, asks.Link, asks.FailedWhat, asks.StandingWhat}, Proofs: []string{asks.CodeProof}},
|
||||
|
||||
@@ -33,13 +33,14 @@ func (h *Holder) Sync(open []Condition, asked time.Time) SyncResult {
|
||||
for _, c := range open {
|
||||
in[c.Key] = true
|
||||
}
|
||||
// A module's condition (Publisher set) is not the controller's to end: only what it said is ended here.
|
||||
// What is open here and no longer there ended while this holder was not told: forgotten without
|
||||
// a message — said only where an edit notifies nobody. What this holder heard of after the
|
||||
// reading was asked for is newer than it, and stands.
|
||||
h.mu.Lock()
|
||||
var gone []*Record
|
||||
for _, r := range h.open {
|
||||
if r.Origin == "condition" && r.Cleared.IsZero() && !in[r.Key] && !r.Seen.After(asked) {
|
||||
if r.Origin == "condition" && r.Publisher == "" && r.Cleared.IsZero() && !in[r.Key] && !r.Seen.After(asked) {
|
||||
gone = append(gone, r)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,22 +1,22 @@
|
||||
module messenger
|
||||
|
||||
go 1.25.0
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.51.0
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
github.com/klauspost/compress v1.18.5 // indirect
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/minio/highwayhash v1.0.4 // indirect
|
||||
github.com/nats-io/jwt/v2 v2.8.1 // indirect
|
||||
github.com/nats-io/nkeys v0.4.15 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.50.0 // indirect
|
||||
golang.org/x/sys v0.43.0 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
golang.org/x/time v0.15.0 // indirect
|
||||
)
|
||||
|
||||
+12
-12
@@ -1,27 +1,27 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 h1:r5xIo5qWOXQdCSMxvqJy8kV4U+P/QWXFK6971Ha6wU8=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 h1:qHpN/fpTDqijfMVoepxUT6mPkfcydb4LiPKsIokv6iM=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
|
||||
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
|
||||
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
|
||||
github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI=
|
||||
github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
|
||||
github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
|
||||
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
|
||||
golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
|
||||
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
|
||||
@@ -16,7 +16,8 @@
|
||||
],
|
||||
"accepts": [
|
||||
"ask",
|
||||
"cancel"
|
||||
"cancel",
|
||||
"condition"
|
||||
],
|
||||
"emits": [
|
||||
"decided"
|
||||
@@ -24,6 +25,7 @@
|
||||
"by-caller": [
|
||||
"ask",
|
||||
"cancel",
|
||||
"condition",
|
||||
"decided"
|
||||
],
|
||||
"records": [
|
||||
|
||||
Reference in New Issue
Block a user