Merge pull request 'The tool runner settles a seat ask with a delay and states seat-settle-delay (issue 1748; group fix/1748-transport-retry)' (#68) from fix/1748-transport-retry into main
This commit was merged in pull request #68.
This commit is contained in:
@@ -147,11 +147,27 @@ const (
|
|||||||
SettleNak = "nak" // handed back, to be delivered again at once
|
SettleNak = "nak" // handed back, to be delivered again at once
|
||||||
SettleTerm = "term" // dropped
|
SettleTerm = "term" // dropped
|
||||||
SettleProgress = "progress" // still being worked: its acknowledgement wait starts again
|
SettleProgress = "progress" // still being worked: its acknowledgement wait starts again
|
||||||
|
// SettleNakAfter hands the message back to be delivered again after a delay: the bus's own redelivery, so no part
|
||||||
|
// keeps a timer for a step given again (novox/hq ADR 0664 §1; issue 1748). SeatSettleAfter gives it.
|
||||||
|
SettleNakAfter = "nak-after"
|
||||||
)
|
)
|
||||||
|
|
||||||
// SeatSettle gives the message module pulled under id its fate. A message pulled for another module, or settled
|
// SeatSettle gives the message module pulled under id its fate. A message pulled for another module, or settled
|
||||||
// already, is refused.
|
// already, is refused.
|
||||||
func (c *Conn) SeatSettle(module, id, fate string) error {
|
func (c *Conn) SeatSettle(module, id, fate string) error {
|
||||||
|
return c.settle(module, id, fate, 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SeatSettleAfter hands the message module pulled under id back to be delivered again after delay (SettleNakAfter),
|
||||||
|
// counting one more delivery. A delay of 0 or below is refused and settles nothing.
|
||||||
|
func (c *Conn) SeatSettleAfter(module, id string, delay time.Duration) error {
|
||||||
|
if delay <= 0 {
|
||||||
|
return fmt.Errorf("a message given again after a delay needs a delay above 0, not %s", delay)
|
||||||
|
}
|
||||||
|
return c.settle(module, id, SettleNakAfter, delay)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Conn) settle(module, id, fate string, delay time.Duration) error {
|
||||||
seatPulls.mu.Lock()
|
seatPulls.mu.Lock()
|
||||||
p, ok := seatPulls.held[id]
|
p, ok := seatPulls.held[id]
|
||||||
if ok && p.module != module {
|
if ok && p.module != module {
|
||||||
@@ -173,9 +189,11 @@ func (c *Conn) SeatSettle(module, id, fate string) error {
|
|||||||
return p.msg.Term()
|
return p.msg.Term()
|
||||||
case SettleProgress:
|
case SettleProgress:
|
||||||
return p.msg.InProgress()
|
return p.msg.InProgress()
|
||||||
|
case SettleNakAfter:
|
||||||
|
return p.msg.NakWithDelay(delay)
|
||||||
}
|
}
|
||||||
seatPulls.mu.Lock()
|
seatPulls.mu.Lock()
|
||||||
seatPulls.held[id] = p // an unknown fate settles nothing
|
seatPulls.held[id] = p // an unknown fate settles nothing
|
||||||
seatPulls.mu.Unlock()
|
seatPulls.mu.Unlock()
|
||||||
return fmt.Errorf("a pulled message's fate is ack, nak, term or progress, not %q", fate)
|
return fmt.Errorf("a pulled message's fate is ack, nak, nak-after, term or progress, not %q", fate)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -58,6 +58,21 @@ func TestAHolderPullsOneMessageWhenItHasRoom(t *testing.T) {
|
|||||||
if err != nil || again == nil || string(again.Body) != `{"id":"a2"}` || again.Delivered != 2 {
|
if err != nil || again == nil || string(again.Body) != `{"id":"a2"}` || again.Delivered != 2 {
|
||||||
t.Fatalf("the message handed back = %+v, %v", again, err)
|
t.Fatalf("the message handed back = %+v, %v", again, err)
|
||||||
}
|
}
|
||||||
|
// Issue 1748: a message settled with a delay is not given again at once, is after it, one delivery higher; a delay
|
||||||
|
// of 0 settles nothing.
|
||||||
|
if err := conn.SeatSettleAfter("messenger", again.ID, 0); err == nil {
|
||||||
|
t.Error("a delay of 0 was accepted")
|
||||||
|
}
|
||||||
|
if err := conn.SeatSettleAfter("messenger", again.ID, 300*time.Millisecond); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if early, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0); err != nil || early != nil {
|
||||||
|
t.Fatalf("a message given again after a delay came at once: %+v, %v", early, err)
|
||||||
|
}
|
||||||
|
late, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 2*time.Second)
|
||||||
|
if err != nil || late == nil || late.Delivered != 3 {
|
||||||
|
t.Fatalf("the message given again after its delay = %+v, %v", late, err)
|
||||||
|
}
|
||||||
if _, err := conn.SeatPull("telegram", "SEAT_OPERATOR_CHANNEL_worker", 0); err == nil || !strings.Contains(err.Error(), "membership lists") {
|
if _, err := conn.SeatPull("telegram", "SEAT_OPERATOR_CHANNEL_worker", 0); err == nil || !strings.Contains(err.Error(), "membership lists") {
|
||||||
t.Errorf("a channel pulled the router's work: %v", err)
|
t.Errorf("a channel pulled the router's work: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -139,7 +139,7 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) {
|
|||||||
t.Fatalf("ping of the instance: %v", err)
|
t.Fatalf("ping of the instance: %v", err)
|
||||||
}
|
}
|
||||||
var ping micro.Ping
|
var ping micro.Ping
|
||||||
if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls" {
|
if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,seat-settle-delay" {
|
||||||
t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err)
|
t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
|
|
||||||
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -25,9 +26,10 @@ var callFamilies = map[string][]string{
|
|||||||
buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag},
|
buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag},
|
||||||
}
|
}
|
||||||
|
|
||||||
// CallCapabilities is every capability of the calls this runner answers, sorted.
|
// CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat
|
||||||
|
// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5).
|
||||||
func CallCapabilities() []string {
|
func CallCapabilities() []string {
|
||||||
out := make([]string, 0, len(callFamilies))
|
out := []string{stdio.CapSeatSettleDelay}
|
||||||
for c := range callFamilies {
|
for c := range callFamilies {
|
||||||
out = append(out, c)
|
out = append(out, c)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) {
|
|||||||
// no other name; none answers "the runtime answers no ...".
|
// no other name; none answers "the runtime answers no ...".
|
||||||
func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) {
|
func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) {
|
||||||
m, _, _ := aLedger(t)
|
m, _, _ := aLedger(t)
|
||||||
if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls" {
|
if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,seat-settle-delay" {
|
||||||
t.Fatalf("the runner states %s", got)
|
t.Fatalf("the runner states %s", got)
|
||||||
}
|
}
|
||||||
for capability, methods := range callFamilies {
|
for capability, methods := range callFamilies {
|
||||||
|
|||||||
@@ -1256,8 +1256,10 @@ type seatAsked struct {
|
|||||||
Bucket string `json:"bucket"`
|
Bucket string `json:"bucket"`
|
||||||
Key string `json:"key"`
|
Key string `json:"key"`
|
||||||
// WaitMS is how long a pull waits for a message (none when 0); Fate what a settle gives one (issue 1887 A11).
|
// WaitMS is how long a pull waits for a message (none when 0); Fate what a settle gives one (issue 1887 A11).
|
||||||
WaitMS int `json:"wait_ms"`
|
WaitMS int `json:"wait_ms"`
|
||||||
Fate string `json:"fate"`
|
// DelayMS is how long a settle with the fate nak-after waits before the bus gives the message again.
|
||||||
|
DelayMS int64 `json:"delay_ms"`
|
||||||
|
Fate string `json:"fate"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// Seat answers a bundle's seat traffic, each as the module and only as far as its membership lists:
|
// Seat answers a bundle's seat traffic, each as the module and only as far as its membership lists:
|
||||||
@@ -1300,7 +1302,13 @@ func (b *moduleBus) Seat(verb string, params json.RawMessage, handOn func(string
|
|||||||
}
|
}
|
||||||
return json.Marshal(p)
|
return json.Marshal(p)
|
||||||
case "settle":
|
case "settle":
|
||||||
if err := conn.SeatSettle(b.module, asked.ID, asked.Fate); err != nil {
|
settle := func() error { return conn.SeatSettle(b.module, asked.ID, asked.Fate) }
|
||||||
|
if asked.Fate == bus.SettleNakAfter {
|
||||||
|
settle = func() error {
|
||||||
|
return conn.SeatSettleAfter(b.module, asked.ID, time.Duration(asked.DelayMS)*time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := settle(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return json.RawMessage(`{}`), nil
|
return json.RawMessage(`{}`), nil
|
||||||
|
|||||||
@@ -202,6 +202,17 @@ func TestABundlePullsAndSettlesThroughItsModulesBus(t *testing.T) {
|
|||||||
if _, err := elsewhere.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err == nil {
|
if _, err := elsewhere.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err == nil {
|
||||||
t.Fatal("another module settled the pulled ask")
|
t.Fatal("another module settled the pulled ask")
|
||||||
}
|
}
|
||||||
|
// Issue 1748: a settle with the fate nak-after and no delay settles nothing; with a delay the ask comes again.
|
||||||
|
if _, err := building.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"nak-after"}`), nil); err == nil {
|
||||||
|
t.Fatal("a delayed nak without a delay was accepted")
|
||||||
|
}
|
||||||
|
if _, err := building.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"nak-after","delay_ms":100}`), nil); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
raw, err = building.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":2000}`), nil)
|
||||||
|
if err != nil || json.Unmarshal(raw, &p) != nil || p.Delivered != 2 {
|
||||||
|
t.Fatalf("the ask given again after its delay = %s, %v", raw, err)
|
||||||
|
}
|
||||||
if _, err := building.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err != nil {
|
if _, err := building.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user