Merge pull request 'A kept call is called again through a transport fault (ADR 0668; issue 1966; group fix/1966-call-made-once, merge first)' (#185) from fix/1966-call-made-once into main
This commit was merged in pull request #185.
This commit is contained in:
+40
-6
@@ -8,10 +8,13 @@
|
||||
// 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.
|
||||
// DoRepeatable is the form for a call its caller keeps the answer of and marks safe to make again (a read, or a write
|
||||
// carrying its own message id): it also asks again on a timeout, and its blackout probe is optional (novox/hq ADR 0668,
|
||||
// extending ADR 0615 §6 to every kept call; issue 1966).
|
||||
package busretry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
@@ -40,7 +43,8 @@ func Waits(bound, first, most time.Duration) (restore func()) {
|
||||
}
|
||||
|
||||
// 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
|
||||
// no response from the stream after the client's own retries on no responders (either JetStream client's: the old
|
||||
// one's publish says it for no responders, review round 1 of issue 1966); also said in words by a runtime that
|
||||
// passed the error on.
|
||||
func Undelivered(err error) bool {
|
||||
var answered *Answered
|
||||
@@ -49,6 +53,7 @@ func Undelivered(err error) bool {
|
||||
}
|
||||
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.ErrNoStreamResponse) ||
|
||||
errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) {
|
||||
return true
|
||||
}
|
||||
@@ -66,8 +71,32 @@ func (a *Answered) Error() string { return a.Words }
|
||||
// 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) {
|
||||
return do(what, try, func(err error) bool { return Undelivered(err) && (blackout == nil || blackout()) })
|
||||
}
|
||||
|
||||
// DoRepeatable is Do for a call the caller marks safe to make again (a read, or a write carrying its message id, which
|
||||
// the callee deduplicates): a timeout is asked again too, and with no blackout probe a request that reached no one is
|
||||
// asked again whether or not the bus is blacked out, since a callee that is restarting answers nothing for seconds
|
||||
// (novox/hq ADR 0668; issue 1966). Past Bound the error says which fault it last met.
|
||||
func DoRepeatable[T any](what string, try func() (T, error), blackout func() bool) (T, error) {
|
||||
return do(what, try, func(err error) bool {
|
||||
return TimedOut(err) || (Undelivered(err) && (blackout == nil || blackout()))
|
||||
})
|
||||
}
|
||||
|
||||
// TimedOut says err is a request's deadline passing without an answer: the request may have been delivered, so only a
|
||||
// call marked safe to make again (DoRepeatable) asks it again. A callee's own answer (Answered) never counts.
|
||||
func TimedOut(err error) bool {
|
||||
var answered *Answered
|
||||
if err == nil || errors.As(err, &answered) {
|
||||
return false
|
||||
}
|
||||
return errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout)
|
||||
}
|
||||
|
||||
func do[T any](what string, try func() (T, error), again func(error) bool) (T, error) {
|
||||
v, err := try()
|
||||
if !Undelivered(err) || (blackout != nil && !blackout()) {
|
||||
if !again(err) {
|
||||
return v, err
|
||||
}
|
||||
start, wait := time.Now(), FirstWait
|
||||
@@ -75,12 +104,17 @@ func Do[T any](what string, try func() (T, error), blackout func() bool) (T, err
|
||||
time.Sleep(wait)
|
||||
wait = min(wait*2, MostWait)
|
||||
v, err = try()
|
||||
if !Undelivered(err) || (blackout != nil && !blackout()) {
|
||||
if !again(err) {
|
||||
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)
|
||||
if TimedOut(err) {
|
||||
err = fmt.Errorf("%s: no answer within its deadline, asked again for %s: %w", what, time.Since(start).Round(time.Second), err)
|
||||
} else {
|
||||
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, a callee that restarts answers nothing; this was longer): %w", what,
|
||||
time.Since(start).Round(time.Second), err)
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "[mesh-sdk] %v\n", err)
|
||||
return v, err
|
||||
}
|
||||
|
||||
@@ -16,30 +16,43 @@ import (
|
||||
// bound; a blackout probe that says the bus answers ends the wait at once.
|
||||
func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) {
|
||||
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
|
||||
name string
|
||||
errs []error // what each try answers; the last repeats
|
||||
blackout func() bool
|
||||
tries int // at least
|
||||
atMost int
|
||||
delivered bool
|
||||
repeatable bool // DoRepeatable, a call marked safe to make again (ADR 0668)
|
||||
}{
|
||||
{"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 }, 2, 1000, false},
|
||||
{"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true, false},
|
||||
{"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true, false},
|
||||
{"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true, false},
|
||||
{"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false, false},
|
||||
{"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false, false},
|
||||
{"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false, false},
|
||||
{"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 2, 1000, false, false},
|
||||
// Issue 1966, ADR 0668: a call kept and safe to make again is asked again on a timeout, and on no responders
|
||||
// from a callee that restarts on a bus that answers; Do still asks neither.
|
||||
{"repeatable: timed out twice, then an answer", []error{context.DeadlineExceeded, nats.ErrTimeout, nil}, nil, 3, 3, true, true},
|
||||
{"repeatable: no responders with no probe, then an answer", []error{nats.ErrNoResponders, nil}, nil, 2, 2, true, true},
|
||||
{"repeatable: a callee's answer is an answer", []error{&Answered{Words: "context deadline exceeded"}}, nil, 1, 1, false, true},
|
||||
{"repeatable: nobody answers and the probe says the bus answers", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false, true},
|
||||
{"repeatable: no response from the stream, either client's, then an answer", []error{nats.ErrNoStreamResponse, jetstream.ErrNoStreamResponse, nil}, nil, 3, 3, true, true},
|
||||
{"timed out past the bound: said", []error{context.DeadlineExceeded}, nil, 2, 1000, false, true},
|
||||
} {
|
||||
// A row that ends in an answer waits on a bound of seconds, which no slow machine reaches; only the row that
|
||||
// runs past it has a short one, and asks only that it was asked again at least once.
|
||||
bound := 10 * time.Second
|
||||
if strings.HasPrefix(row.name, "blacked out") {
|
||||
if strings.Contains(row.name, "past the bound") {
|
||||
bound = 100 * time.Millisecond
|
||||
}
|
||||
t.Cleanup(Waits(bound, time.Millisecond, 4*time.Millisecond))
|
||||
tries := 0
|
||||
_, err := Do("the test", func() (int, error) {
|
||||
do := Do[int]
|
||||
if row.repeatable {
|
||||
do = DoRepeatable[int]
|
||||
}
|
||||
_, err := do("the test", func() (int, error) {
|
||||
e := row.errs[min(tries, len(row.errs)-1)]
|
||||
tries++
|
||||
return tries, e
|
||||
@@ -50,5 +63,8 @@ func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) {
|
||||
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)
|
||||
}
|
||||
if strings.HasPrefix(row.name, "timed out") && (!TimedOut(err) || !strings.Contains(fmt.Sprint(err), "the test: no answer within its deadline")) {
|
||||
t.Errorf("%s: %v", row.name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+52
-7
@@ -51,6 +51,57 @@ func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) {
|
||||
return c.callOn(called, fallback, body)
|
||||
}
|
||||
|
||||
// CallKept is Call for a call whose answer the caller keeps and which is safe to make again (a read, or a write the
|
||||
// callee deduplicates): it is asked again through a transport fault, a timeout or no responders, whether or not the bus
|
||||
// is blacked out, until it is answered or busretry.Bound has passed (novox/hq ADR 0668, extending ADR 0615 §6; issue
|
||||
// 1966). A callee that restarts answers no one for seconds; Call fails that at once. The caller keeps a call still
|
||||
// unanswered past the bound as a wait of its own, and makes it again on the callee's event or its consumer's bind.
|
||||
func (c *Conn) CallKept(key string, body any) (json.RawMessage, error) {
|
||||
called, fallback, err := callSubjectsOn(key, "", c.callerToken())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.callKeptOn(called, fallback, body)
|
||||
}
|
||||
|
||||
// CallKeptOn is CallKept to the tool or seat verb of one named machine, as CallOn is Call's.
|
||||
func (c *Conn) CallKeptOn(node, key string, body any) (json.RawMessage, error) {
|
||||
if err := calladdress.CheckNode(node); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
called, fallback, err := callSubjectsOn(key, node, c.callerToken())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.callKeptOn(called, fallback, body)
|
||||
}
|
||||
|
||||
func (c *Conn) callKeptOn(called, fallback string, body any) (json.RawMessage, error) {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return callKeptWith(c.request, called, fallback, data)
|
||||
}
|
||||
|
||||
// callKeptWith is callWith for a kept call: busretry.DoRepeatable with no blackout probe.
|
||||
func callKeptWith(request func(string, []byte) (json.RawMessage, error), called, fallback string,
|
||||
data []byte) (json.RawMessage, error) {
|
||||
return busretry.DoRepeatable("the call on "+fallback, askBoth(request, called, fallback, data), nil)
|
||||
}
|
||||
|
||||
// askBoth asks called, then fallback when nothing answers it.
|
||||
func askBoth(request func(string, []byte) (json.RawMessage, error), called, fallback string,
|
||||
data []byte) func() (json.RawMessage, error) {
|
||||
return func() (json.RawMessage, error) {
|
||||
result, err := request(called, data)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
result, err = request(fallback, data)
|
||||
}
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -67,13 +118,7 @@ func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error
|
||||
// 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)
|
||||
return busretry.Do("the call on "+fallback, askBoth(request, called, fallback, data), blackout)
|
||||
}
|
||||
|
||||
// answered is the error a callee answered with: delivered, never asked again (busretry.Answered).
|
||||
|
||||
+17
-5
@@ -1,6 +1,7 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -108,17 +109,28 @@ func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) {
|
||||
name string
|
||||
answer []error // per ask; the last repeats
|
||||
asks int
|
||||
kept bool // CallKept (novox/hq ADR 0668; issue 1966)
|
||||
}{
|
||||
{"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},
|
||||
{"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1, false},
|
||||
{"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1, false},
|
||||
{"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3, false},
|
||||
{"a timeout may have been delivered: Call asks once", []error{context.DeadlineExceeded, nil}, 1, false},
|
||||
{"kept: no responders on a bus that answers (a callee restarting), then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3, true},
|
||||
{"kept: a timeout, then an answer", []error{context.DeadlineExceeded, nil}, 2, true},
|
||||
{"kept: a callee's answer is asked once", []error{answered("nats: timeout")}, 1, true},
|
||||
} {
|
||||
asks := 0
|
||||
_, err := callWith(func(string, []byte) (json.RawMessage, error) {
|
||||
request := 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)
|
||||
}
|
||||
var err error
|
||||
if row.kept {
|
||||
_, err = callKeptWith(request, "called", "fallback", nil)
|
||||
} else {
|
||||
_, err = callWith(request, 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)
|
||||
}
|
||||
|
||||
@@ -59,7 +59,9 @@ func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
|
||||
|
||||
// OnBound registers f, told each time the hub binds the module's consumer, with when the bus made it. A consumer is made
|
||||
// from now (ADR 0443), so an event said before it was made is not delivered to it: a part that took a picture before
|
||||
// that must ask its owner again (views.Follower.ConsumerMade). Register before Start.
|
||||
// that must ask its owner again (views.Follower.ConsumerMade). So every handler of an event that ends a kept call's wait
|
||||
// (a callee's "my state changed") also registers here the call it makes on that event, made again once at each bind
|
||||
// (novox/hq ADR 0668; issue 1966). Register before Start.
|
||||
func (h *EventHub) OnBound(f func(created time.Time)) {
|
||||
h.mu.Lock()
|
||||
h.bound = append(h.bound, f)
|
||||
|
||||
@@ -43,7 +43,7 @@
|
||||
},
|
||||
{
|
||||
"name": "presence",
|
||||
"description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.",
|
||||
"description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.",
|
||||
"input": {
|
||||
"properties": {
|
||||
"machine": {
|
||||
@@ -55,7 +55,54 @@
|
||||
},
|
||||
"answer": {
|
||||
"type": "object",
|
||||
"description": "who is connected and who is silent"
|
||||
"description": "with machine, that machine: online and since; without, every key of the state presence",
|
||||
"properties": {
|
||||
"machine": {
|
||||
"type": "string",
|
||||
"description": "the machine asked, with machine"
|
||||
},
|
||||
"online": {
|
||||
"type": "boolean",
|
||||
"description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of"
|
||||
},
|
||||
"since": {
|
||||
"type": "string",
|
||||
"description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "without machine: the state read, presence"
|
||||
},
|
||||
"holds": {
|
||||
"type": "integer",
|
||||
"description": "without machine: how many keys the state holds"
|
||||
},
|
||||
"keys": {
|
||||
"type": "array",
|
||||
"description": "without machine: each key of the state presence, <node> or part.<part>.<instance>, with its value",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<node> or part.<part>.<instance>"
|
||||
},
|
||||
"value": {
|
||||
"type": "object",
|
||||
"description": "the key's value, online and its layout's version"
|
||||
},
|
||||
"revision": {
|
||||
"type": "integer",
|
||||
"description": "the key's revision"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"key",
|
||||
"value"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
|
||||
@@ -116,15 +116,65 @@
|
||||
},
|
||||
{
|
||||
"name": "asks",
|
||||
"description": "The asks the router holds: open first, then the last month's, each with how it ended.",
|
||||
"description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.",
|
||||
"input": {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"description": "no arguments"
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "one ask, by its key (<asker>.<id>); none for the open asks"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "open (open and held), ended or all; none is open when no key or since is given, else all"
|
||||
},
|
||||
"since": {
|
||||
"type": "string",
|
||||
"description": "only the asks opened at or after this time, RFC 3339"
|
||||
},
|
||||
"bound": {
|
||||
"type": "integer",
|
||||
"description": "the most bytes of the asks in the answer; said when it cuts the answer short"
|
||||
}
|
||||
}
|
||||
},
|
||||
"answer": {
|
||||
"type": "object",
|
||||
"description": "the asks"
|
||||
"description": "the asks, open first",
|
||||
"properties": {
|
||||
"asks": {
|
||||
"type": "array",
|
||||
"description": "each ask the router holds, as its state asks holds it",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<asker>.<id>"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"key",
|
||||
"state"
|
||||
]
|
||||
}
|
||||
},
|
||||
"cut": {
|
||||
"type": "string",
|
||||
"description": "what the bound cut, and how to ask for more; empty when nothing was cut"
|
||||
},
|
||||
"matched": {
|
||||
"type": "integer",
|
||||
"description": "how many asks the arguments matched"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"asks"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
// one seat (that definition, Answer); the seat's two events, whose bodies name the seat (Changed). It calls the seat,
|
||||
// never the module that holds it, and only trunk definitions are held: a pull request's are the indexer's to judge.
|
||||
// An ask that reached no one (busretry.Undelivered: the bus's "no responders", as while the seat's holder re-serves) is
|
||||
// asked again within the bus's retry bound (busretry.Bound, about 30 s, backing off; issue 774's form, issue 1966).
|
||||
// asked again within the bus's retry bound (busretry.Bound, about 30 s, backing off; issue 774's form, issue 1966), and so is one
|
||||
// that timed out: the ask is a read, safe to make again (busretry.DoRepeatable; novox/hq ADR 0668).
|
||||
// Nothing else polls or waits on a timer: a reader that could not ask says so in its Status and reads again on the
|
||||
// next event or on Reload, which the host offers as a tool (Tools).
|
||||
//
|
||||
@@ -361,11 +362,12 @@ func (r *Reader) ReloadIfUnread() <-chan struct{} {
|
||||
return done
|
||||
}
|
||||
|
||||
// ask asks the seat's verb, and again while the ask reached no one, within the bus's retry bound: a holder that re-serves
|
||||
// for a few seconds while the reader starts leaves no list unread (issue 1966). An answer, a refusal or a timeout is never
|
||||
// asked again.
|
||||
// ask asks the seat's verb, and again while the ask reached no one or timed out, within the bus's retry bound: a holder
|
||||
// that re-serves for a few seconds while the reader starts leaves no list unread (issue 1966). The ask is a read, so it
|
||||
// is safe to make again on a timeout (busretry.DoRepeatable; novox/hq ADR 0668). An answer or a refusal is never asked
|
||||
// again.
|
||||
func (r *Reader) ask(body map[string]any, what string) (json.RawMessage, error) {
|
||||
return busretry.Do(what, func() (json.RawMessage, error) {
|
||||
return busretry.DoRepeatable(what, func() (json.RawMessage, error) {
|
||||
return r.cfg.Bus.Ask(Key(), body)
|
||||
}, nil)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package seatlist
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -22,6 +23,7 @@ type fakeOwner struct {
|
||||
table map[string]string // seat -> definition JSON, trunk
|
||||
down bool
|
||||
silent int // when set, so many asks first reach no one (the bus's no responders), as while the holder re-serves
|
||||
timeouts int // when set, so many asks first time out (a deadline passed, as a callee that restarts mid-ask)
|
||||
gate chan struct{} // when set, an ask waits for it to be closed
|
||||
asked []string // "key seat"
|
||||
handlers map[string]func(string, json.RawMessage) error
|
||||
@@ -34,6 +36,10 @@ func (f *fakeOwner) Ask(key string, body any) (json.RawMessage, error) {
|
||||
args, _ := body.(map[string]any)
|
||||
seat, _ := args["seat"].(string)
|
||||
f.asked = append(f.asked, key+" "+seat)
|
||||
if f.timeouts > 0 {
|
||||
f.timeouts--
|
||||
return nil, fmt.Errorf("asking %s: %w", key, context.DeadlineExceeded)
|
||||
}
|
||||
if f.silent > 0 {
|
||||
f.silent--
|
||||
return nil, fmt.Errorf("not asked: nothing on this machine answers it (%w)", nats.ErrNoResponders)
|
||||
@@ -287,6 +293,17 @@ func TestStartReturnsAtOnceAndTheReaderSaysItIsReadingUntilTheSeatAnswers(t *tes
|
||||
if st := r.Status(); !st.Heard || st.LastError != "" || !r.Has("node-b") || len(owner.asked) != 4 {
|
||||
t.Fatalf("after a holder that re-served: %+v, asked %q: want heard on the third ask of the list", st, owner.asked)
|
||||
}
|
||||
|
||||
// Issue 1966, ADR 0668: a read of the list is safe to make again, so an ask that timed out is asked again too.
|
||||
owner = &fakeOwner{timeouts: 1, table: map[string]string{"node-b": def("node-b", "node", "put")}}
|
||||
r, err = Start(Config{Bus: owner, Bootstrap: Bootstrapped()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
<-r.Ready()
|
||||
if st := r.Status(); !st.Heard || st.LastError != "" || !r.Has("node-b") || len(owner.asked) != 3 {
|
||||
t.Fatalf("after an ask that timed out: %+v, asked %q: want heard on the second ask of the list", st, owner.asked)
|
||||
}
|
||||
}
|
||||
|
||||
// The bootstrap and the tests' seat list are copies of the same contracts files: where a seat is in both, they are equal,
|
||||
|
||||
@@ -43,7 +43,7 @@
|
||||
},
|
||||
{
|
||||
"name": "presence",
|
||||
"description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.",
|
||||
"description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.",
|
||||
"input": {
|
||||
"properties": {
|
||||
"machine": {
|
||||
@@ -55,7 +55,54 @@
|
||||
},
|
||||
"answer": {
|
||||
"type": "object",
|
||||
"description": "who is connected and who is silent"
|
||||
"description": "with machine, that machine: online and since; without, every key of the state presence",
|
||||
"properties": {
|
||||
"machine": {
|
||||
"type": "string",
|
||||
"description": "the machine asked, with machine"
|
||||
},
|
||||
"online": {
|
||||
"type": "boolean",
|
||||
"description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of"
|
||||
},
|
||||
"since": {
|
||||
"type": "string",
|
||||
"description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "without machine: the state read, presence"
|
||||
},
|
||||
"holds": {
|
||||
"type": "integer",
|
||||
"description": "without machine: how many keys the state holds"
|
||||
},
|
||||
"keys": {
|
||||
"type": "array",
|
||||
"description": "without machine: each key of the state presence, <node> or part.<part>.<instance>, with its value",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<node> or part.<part>.<instance>"
|
||||
},
|
||||
"value": {
|
||||
"type": "object",
|
||||
"description": "the key's value, online and its layout's version"
|
||||
},
|
||||
"revision": {
|
||||
"type": "integer",
|
||||
"description": "the key's revision"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"key",
|
||||
"value"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
|
||||
@@ -116,15 +116,65 @@
|
||||
},
|
||||
{
|
||||
"name": "asks",
|
||||
"description": "The asks the router holds: open first, then the last month's, each with how it ended.",
|
||||
"description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.",
|
||||
"input": {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"description": "no arguments"
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "one ask, by its key (<asker>.<id>); none for the open asks"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "open (open and held), ended or all; none is open when no key or since is given, else all"
|
||||
},
|
||||
"since": {
|
||||
"type": "string",
|
||||
"description": "only the asks opened at or after this time, RFC 3339"
|
||||
},
|
||||
"bound": {
|
||||
"type": "integer",
|
||||
"description": "the most bytes of the asks in the answer; said when it cuts the answer short"
|
||||
}
|
||||
}
|
||||
},
|
||||
"answer": {
|
||||
"type": "object",
|
||||
"description": "the asks"
|
||||
"description": "the asks, open first",
|
||||
"properties": {
|
||||
"asks": {
|
||||
"type": "array",
|
||||
"description": "each ask the router holds, as its state asks holds it",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<asker>.<id>"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"key",
|
||||
"state"
|
||||
]
|
||||
}
|
||||
},
|
||||
"cut": {
|
||||
"type": "string",
|
||||
"description": "what the bound cut, and how to ask for more; empty when nothing was cut"
|
||||
},
|
||||
"matched": {
|
||||
"type": "integer",
|
||||
"description": "how many asks the arguments matched"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"asks"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
|
||||
Vendored
+1
@@ -12,6 +12,7 @@ repositories; before that it was mesh-controller's copy, which still held two mo
|
||||
|
||||
mesh-catalog b5dbfbb784507e211372a7c9640d91f7d6dadd84 modules/*/module.json, modules/*/files/**, modules/*/templates/**, predefined.json
|
||||
mesh-contracts 70fe186d7f5743e61f3ce2bbd97a8bfabc5b596e the seat files of the eight bootstrapped seats held there (seats/, health-and-repair/seats/, operator-and-conversation/seats/), as seat.json.captured
|
||||
mesh-contracts bb16f99e2d5e854bb8f6552ce4b256e7c10e8ea0 of those, health-and-repair/seats/mesh-watchdogs and operator-and-conversation/seats/operator-channel (issue 1966)
|
||||
mesh-delivery 932e7ccf025513ec4fce38a534c2c38defbc6701 contracts/seats/mesh-catalog and mesh-delivery-checker, as seat.json.captured (go/seatlist's bootstrap test)
|
||||
mesh-host cdcba9ca73755854d21e4e657d7e46c32b9b5265 internal/apply/placement_guard.go
|
||||
|
||||
|
||||
Vendored
+49
-2
@@ -43,7 +43,7 @@
|
||||
},
|
||||
{
|
||||
"name": "presence",
|
||||
"description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.",
|
||||
"description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.",
|
||||
"input": {
|
||||
"properties": {
|
||||
"machine": {
|
||||
@@ -55,7 +55,54 @@
|
||||
},
|
||||
"answer": {
|
||||
"type": "object",
|
||||
"description": "who is connected and who is silent"
|
||||
"description": "with machine, that machine: online and since; without, every key of the state presence",
|
||||
"properties": {
|
||||
"machine": {
|
||||
"type": "string",
|
||||
"description": "the machine asked, with machine"
|
||||
},
|
||||
"online": {
|
||||
"type": "boolean",
|
||||
"description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of"
|
||||
},
|
||||
"since": {
|
||||
"type": "string",
|
||||
"description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "without machine: the state read, presence"
|
||||
},
|
||||
"holds": {
|
||||
"type": "integer",
|
||||
"description": "without machine: how many keys the state holds"
|
||||
},
|
||||
"keys": {
|
||||
"type": "array",
|
||||
"description": "without machine: each key of the state presence, <node> or part.<part>.<instance>, with its value",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<node> or part.<part>.<instance>"
|
||||
},
|
||||
"value": {
|
||||
"type": "object",
|
||||
"description": "the key's value, online and its layout's version"
|
||||
},
|
||||
"revision": {
|
||||
"type": "integer",
|
||||
"description": "the key's revision"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"key",
|
||||
"value"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
|
||||
+54
-4
@@ -116,15 +116,65 @@
|
||||
},
|
||||
{
|
||||
"name": "asks",
|
||||
"description": "The asks the router holds: open first, then the last month's, each with how it ended.",
|
||||
"description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.",
|
||||
"input": {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"description": "no arguments"
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "one ask, by its key (<asker>.<id>); none for the open asks"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "open (open and held), ended or all; none is open when no key or since is given, else all"
|
||||
},
|
||||
"since": {
|
||||
"type": "string",
|
||||
"description": "only the asks opened at or after this time, RFC 3339"
|
||||
},
|
||||
"bound": {
|
||||
"type": "integer",
|
||||
"description": "the most bytes of the asks in the answer; said when it cuts the answer short"
|
||||
}
|
||||
}
|
||||
},
|
||||
"answer": {
|
||||
"type": "object",
|
||||
"description": "the asks"
|
||||
"description": "the asks, open first",
|
||||
"properties": {
|
||||
"asks": {
|
||||
"type": "array",
|
||||
"description": "each ask the router holds, as its state asks holds it",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"key": {
|
||||
"type": "string",
|
||||
"description": "<asker>.<id>"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"key",
|
||||
"state"
|
||||
]
|
||||
}
|
||||
},
|
||||
"cut": {
|
||||
"type": "string",
|
||||
"description": "what the bound cut, and how to ask for more; empty when nothing was cut"
|
||||
},
|
||||
"matched": {
|
||||
"type": "integer",
|
||||
"description": "how many asks the arguments matched"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"asks"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user