116 lines
4.5 KiB
Go
116 lines
4.5 KiB
Go
// Command mesh-delivery-teller-tools serves the teller's recovery tool (issue 774): the runtime launches it for a call.
|
|
// teller_end_ask writes one value of the teller's state ends and waits for the teller's answer there. The teller is the
|
|
// other program: it serves no request subject (ADR 0437 §4).
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"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-sdk/go/views"
|
|
|
|
"mesh-delivery-teller/internal/ends"
|
|
)
|
|
|
|
// asker writes an ends value and waits for the teller's answer to it.
|
|
type asker func(key string, value []byte) (ends.End, error)
|
|
|
|
// Tools are the teller's tools.
|
|
func Tools(ask asker, now func() time.Time) []stdio.Tool {
|
|
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
|
|
return []stdio.Tool{
|
|
{Name: ends.Tool,
|
|
Description: "End one open ask the delivery teller holds at the operator channel, by hand, saying why: the teller " +
|
|
"cancels it at the router, which frees one of the three places the router gives the teller, or drops it from the " +
|
|
"teller's wait for room. For an ask whose question is gone but whose key was never taken away (a merge ask of a pull " +
|
|
"request merged by another path, a backup ask about a backup no longer failing), which otherwise holds its place " +
|
|
"until it lapses, so a newer ask is refused for want of room. Name the ask by its router id (as " +
|
|
"operator-channel's asks show it) or by the key of the gate's or the assigner's asks. Answers which ask it found, " +
|
|
"whether it ended it, and what it did; an ask answered, lapsed or ended already is left and said so.",
|
|
Input: map[string]any{"type": "object", "required": []string{"ask", "why"}, "properties": map[string]any{
|
|
"ask": map[string]any{"type": "string", "description": "the ask's router id, or the key of the gate's or the assigner's asks it was asked for"},
|
|
"why": map[string]any{"type": "string", "description": "why it is ended by hand, in plain words"}}},
|
|
Run: func(args map[string]any) (any, error) {
|
|
key, value, err := ends.New(str(args, "ask"), str(args, "why"), stdio.CallerOf(args), now())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
e, err := ask(key, value)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return e.Result, nil
|
|
}},
|
|
}
|
|
}
|
|
|
|
// askEnd writes the ends value at key and waits, on a watch of the key, at most bound for the teller's answer.
|
|
func askEnd(s stdio.KeptState, bound time.Duration) asker {
|
|
return func(key string, value []byte) (ends.End, error) {
|
|
answered := make(chan ends.End, 1)
|
|
var rev uint64
|
|
var mu sync.Mutex
|
|
take := func(revision uint64, v []byte) {
|
|
mu.Lock()
|
|
mine := rev
|
|
mu.Unlock()
|
|
if mine == 0 || revision <= mine {
|
|
return
|
|
}
|
|
e, err := ends.Decode(v)
|
|
if err != nil || !e.Done() {
|
|
return
|
|
}
|
|
select {
|
|
case answered <- e:
|
|
default:
|
|
}
|
|
}
|
|
if err := s.Watch(key, func(c stdio.StateChange) error {
|
|
if c.Op == "put" {
|
|
take(c.Revision, c.Value)
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
return ends.End{}, fmt.Errorf("watching %s.%s: %w", ends.State, key, err)
|
|
}
|
|
r, err := s.Put(key, json.RawMessage(value))
|
|
if err != nil {
|
|
return ends.End{}, fmt.Errorf("the end could not be written to %s: %w", ends.Bucket, err)
|
|
}
|
|
mu.Lock()
|
|
rev = r
|
|
mu.Unlock()
|
|
// The watch may have heard the answer before the revision was known: read it once too.
|
|
if e, err := s.Get(key); err == nil && e != nil {
|
|
take(e.Revision, e.Value)
|
|
}
|
|
wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8)
|
|
defer cancel()
|
|
select {
|
|
case e := <-answered:
|
|
return e, nil
|
|
case <-wait.Done():
|
|
return ends.End{}, fmt.Errorf("the teller did not answer within %s: is it running? its answer will be in %s.%s", bound, ends.Bucket, key)
|
|
}
|
|
}
|
|
}
|
|
|
|
func main() {
|
|
// teller_first_views and teller_first_view_again: where the teller's first view of the assigner's asks stands, and
|
|
// asking it again by hand (ADR 0615, design 25 § 12.2).
|
|
tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now),
|
|
views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
|
|
if err := stdio.Serve("", tools); err != nil {
|
|
fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|