diff --git a/README.md b/README.md index 22d54b8..45b8607 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,11 @@ compare-and-set. The runtime does each only as far as the module's membership li contract on top: the `Ask` an asker publishes and the checks it is held to, the `Warrant` it acts on (`For`, `Says`), and the `Message`, `Choice`, `LinkAsked` and `Code` a channel takes and says, with the subjects each travels on. +`go/asker` is an asking module's client (ADR 0259 §10): `Ask` keeps an ask in the module's own state and then +publishes it under the module's name; `Cancel`; `Where` reads its state from the router's record; `Listen` and +`Take` hand the module each warrant once — marked acted on by compare-and-set before the module acts, and only +for an ask it made, an option it offered, at that option's level, before it expired. A lost compare-and-set is +`stdio.ErrChanged`, told by the runtime's error code. ### `primitives` — sealing, semver, resolved-env Seal/unseal for secrets, semantic version comparison, and the accessor a module uses to read its diff --git a/go/asker/asker.go b/go/asker/asker.go new file mode 100644 index 0000000..360c7b6 --- /dev/null +++ b/go/asker/asker.go @@ -0,0 +1,196 @@ +// Package asker is a module's client for asking a person (novox/hq ADR 0259 §10): it publishes an ask under +// the module's own name, cancels it, reads its state from the router's record, and hands the module each +// warrant once — only one for an ask it made, for an option it offered, at that option's level, before it +// expired. The module then acts with its own grants. +// +// It keeps what it asked, and which warrants it acted on, in a state of the module's own (declared in its +// manifest, by default `asked`): the ask is written before it is published, and a warrant is marked acted on +// by compare-and-set before the module's handler runs, so a redelivery, or a second instance, acts once. +package asker + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// Client asks as one module. +type Client struct { + // Module is the asking module's name: the one the bus grants its asks under. + Module string + // State is the module's own state the asks are kept in. + State Store + // Now is the clock; nil is time.Now. + Now func() time.Time + // Publish, Record and Subscribe reach the bus; nil is the stdio SDK's, through the module's runtime. + Publish func(subject string, body any, id string) error + Record func(key string) (*stdio.StateEntry, error) + Subscribe func(pattern string, handle func(stdio.Envelope) error) error +} + +// Store is where the asks are kept: stdio.State(name) is one. +type Store interface { + Get(key string) (*stdio.StateEntry, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) +} + +// New is a client for the module the runtime serves, keeping its asks in the state named. +func New(state string) *Client { + return &Client{Module: os.Getenv("MESH_MODULE"), State: stdio.State(state)} +} + +func (c *Client) now() time.Time { + if c.Now != nil { + return c.Now() + } + return time.Now() +} + +// Kept is one ask as the module keeps it. +type Kept struct { + Ask asks.Ask `json:"ask"` + Asked time.Time `json:"asked"` + Acted bool `json:"acted,omitempty"` + Ended asks.Outcome `json:"ended,omitempty"` + Answer *asks.Warrant `json:"answer,omitempty"` +} + +func key(id string) string { return "ask." + id } + +// Ask checks an ask, keeps it, and publishes it under the module's own name. An ask with an id already kept +// is refused: a new question is a new id. +func (c *Client) Ask(a asks.Ask) error { + if err := a.Check(c.now()); err != nil { + return err + } + if _, err := c.State.Create(key(a.ID), Kept{Ask: a, Asked: c.now()}); err != nil { + if errors.Is(err, stdio.ErrChanged) { + return fmt.Errorf("an ask %s was already made: a new question is a new id", a.ID) + } + return err + } + return c.publish(asks.AskSubject(c.Module), a, a.ID) +} + +// Cancel takes an ask back: the router ends it and edits every copy. +func (c *Client) Cancel(id string) error { + return c.publish(asks.CancelSubject(c.Module), map[string]string{"id": id}, "cancel."+id) +} + +func (c *Client) publish(subject string, body any, id string) error { + if c.Publish != nil { + return c.Publish(subject, body, id) + } + return stdio.SeatPublish(subject, body, id) +} + +// Where is an ask's state as the router keeps it: its state and its warrant, if any — read under the +// module's own name and no other's. +func (c *Client) Where(id string) (state string, w *asks.Warrant, err error) { + read := c.Record + if read == nil { + read = func(k string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", k) } + } + e, err := read(id) + if err != nil || e == nil { + return "", nil, err + } + var rec struct { + State string `json:"state"` + Warrant *asks.Warrant `json:"warrant"` + } + if err := json.Unmarshal(e.Value, &rec); err != nil { + return "", nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err) + } + return rec.State, rec.Warrant, nil +} + +// Handled is what came of a warrant: acted on, refused (and why), or already done. +type Handled string + +const ( + Acted Handled = "acted" + Refused Handled = "refused" + Done Handled = "already acted on" + Recorded Handled = "ended without a choice" +) + +// Take hands act a warrant once, after holding it to the ask it answers. act performs the option with the +// module's own grants; its error is said and the warrant stays acted on (a person's decision is not +// performed twice; the module says the failure in its own record). +func (c *Client) Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (Handled, error) { + if w.Asker != c.Module { + return Refused, fmt.Errorf("a warrant for %s reached %s", w.Asker, c.Module) + } + e, err := c.State.Get(key(w.Ask)) + if err != nil { + return "", err + } + if e == nil { + return Refused, fmt.Errorf("%s holds no ask %s", c.Module, w.Ask) + } + var k Kept + if err := json.Unmarshal(e.Value, &k); err != nil { + return "", err + } + if k.Acted || k.Ended != "" { + return Done, nil + } + if w.Outcome != asks.OutcomeChosen { + k.Ended = w.Outcome + k.Answer = &w + if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil && !errors.Is(err, stdio.ErrChanged) { + return "", err + } + return Recorded, nil + } + o, err := w.For(c.Module, k.Ask) + if err != nil { + return Refused, err + } + // Marked before acting: of two deliveries, or two instances, only the one whose write stands acts. + k.Acted, k.Ended, k.Answer = true, asks.OutcomeChosen, &w + if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil { + if errors.Is(err, stdio.ErrChanged) { + return Done, nil + } + return "", err + } + return Acted, act(o, w) +} + +// Listen hands every warrant for this module to Take, with act. It asks the runtime for the module's +// events matching the router's `decided` subject for this module. +func (c *Client) Listen(act func(asks.Option, asks.Warrant) error, said func(Handled, asks.Warrant, error)) error { + subscribe := c.Subscribe + if subscribe == nil { + subscribe = stdio.Subscribe + } + return subscribe(asks.Seat+".decided."+c.Module, func(e stdio.Envelope) error { + if !strings.HasPrefix(e.Key, asks.Seat+".decided.") { + return nil + } + var w asks.Warrant + if err := json.Unmarshal(e.Body, &w); err != nil { + if said != nil { + said(Refused, w, fmt.Errorf("a warrant could not be read: %w", err)) + } + return nil + } + h, err := c.Take(w, act) + if h == "" && err != nil { + return err // the state could not be reached: offered again + } + if said != nil { + said(h, w, err) + } + return nil + }) +} diff --git a/go/asker/asker_test.go b/go/asker/asker_test.go new file mode 100644 index 0000000..4550784 --- /dev/null +++ b/go/asker/asker_test.go @@ -0,0 +1,130 @@ +package asker + +import ( + "encoding/json" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +type mem struct { + vals map[string]json.RawMessage + revs map[string]uint64 + n uint64 +} + +func (m *mem) Get(k string) (*stdio.StateEntry, error) { + v, ok := m.vals[k] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: k, Value: v, Revision: m.revs[k]}, nil +} +func (m *mem) Create(k string, v any) (uint64, error) { + if _, ok := m.vals[k]; ok { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *mem) Update(k string, v any, rev uint64) (uint64, error) { + if rev == 0 || m.revs[k] != rev { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *mem) put(k string, v any) (uint64, error) { + raw, _ := json.Marshal(v) + m.n++ + m.vals[k], m.revs[k] = raw, m.n + return m.n, nil +} + +var now = time.Date(2026, 10, 8, 14, 0, 0, 0, time.UTC) + +func client() (*Client, map[string][]string) { + sent := map[string][]string{} + c := &Client{Module: "mesh-delivery", State: &mem{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}}, + Now: func() time.Time { return now }, + Publish: func(subject string, body any, id string) error { + sent[subject] = append(sent[subject], id) + return nil + }} + return c, sent +} + +func held() asks.Ask { + return asks.Ask{ID: "d-1", Headline: "Delivery of hq held", Explanation: "Needs you: release it, or stop it.", + Who: asks.Operator, Expires: now.Add(time.Hour), OnExpiry: "the delivery stays held", + Options: []asks.Option{{ID: "release", Label: "Release", Does: "it goes on", Level: asks.Approve}}} +} + +func warrant() asks.Warrant { + return asks.Warrant{Ask: "d-1", Asker: "mesh-delivery", Outcome: asks.OutcomeChosen, Option: "release", + Label: "Release", Level: asks.Approve, At: now.Add(time.Minute), + By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}} +} + +func TestAnAskIsKeptThenPublishedUnderTheModulesOwnName(t *testing.T) { + c, sent := client() + if err := c.Ask(held()); err != nil { + t.Fatal(err) + } + if got := sent[asks.AskSubject("mesh-delivery")]; len(got) != 1 || got[0] != "d-1" { + t.Fatalf("%v", sent) + } + if err := c.Ask(held()); err == nil || !strings.Contains(err.Error(), "already made") { + t.Errorf("the same id was asked twice: %v", err) + } + bad := held() + bad.Who = "agent" + if err := c.Ask(bad); err == nil { + t.Error("an ask the contract refuses was published") + } +} + +func TestAWarrantIsActedOnOnceAndOnlyForItsOwnAsk(t *testing.T) { + c, _ := client() + _ = c.Ask(held()) + acted := 0 + act := func(asks.Option, asks.Warrant) error { acted++; return nil } + if h, err := c.Take(warrant(), act); h != Acted || err != nil || acted != 1 { + t.Fatalf("%s %v %d", h, err, acted) + } + if h, _ := c.Take(warrant(), act); h != Done || acted != 1 { + t.Errorf("a redelivery was acted on: %s", h) + } + for why, change := range map[string]func(*asks.Warrant){ + "another asker": func(w *asks.Warrant) { w.Asker = "mesh-controller" }, + "an ask not made": func(w *asks.Warrant) { w.Ask = "d-9" }, + "another level": func(w *asks.Warrant) { w.Level = asks.Acknowledge }, + "an option not offered": func(w *asks.Warrant) { w.Option = "delete" }, + "after its expiry": func(w *asks.Warrant) { w.At = now.Add(2 * time.Hour) }, + } { + c2, _ := client() + _ = c2.Ask(held()) + w := warrant() + change(&w) + if h, _ := c2.Take(w, act); h == Acted { + t.Errorf("a warrant from %s was acted on", why) + } + } + if acted != 1 { + t.Errorf("acted %d times", acted) + } +} + +func TestAnAskEndedWithoutAChoiceIsRecordedAndNothingDone(t *testing.T) { + c, _ := client() + _ = c.Ask(held()) + w := warrant() + w.Outcome, w.Option, w.By = asks.OutcomeExpired, "", nil + if h, err := c.Take(w, func(asks.Option, asks.Warrant) error { t.Fatal("acted"); return nil }); h != Recorded || err != nil { + t.Fatalf("%s %v", h, err) + } + if h, _ := c.Take(warrant(), func(asks.Option, asks.Warrant) error { t.Fatal("acted after it ended"); return nil }); h != Done { + t.Errorf("%s", h) + } +} diff --git a/go/asks/asks.go b/go/asks/asks.go index 52dfe16..491d2ba 100644 --- a/go/asks/asks.go +++ b/go/asks/asks.go @@ -289,6 +289,11 @@ func (w Warrant) For(asker string, a Ask) (Option, error) { if w.Level != o.Level { return Option{}, fmt.Errorf("the option %s is %s, and the warrant says %s", o.ID, o.Level, w.Level) } + // A choice made after the ask expired is no answer to it, whatever the router said. + if !w.At.IsZero() && w.At.After(a.Expires) { + return Option{}, fmt.Errorf("the warrant was given at %s, after the ask %s expired at %s", + w.At.UTC().Format(time.RFC3339), a.ID, a.Expires.UTC().Format(time.RFC3339)) + } return o, nil } diff --git a/go/stdio.go b/go/stdio.go index d0cc09b..3582f14 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -48,6 +48,7 @@ type message struct { Params json.RawMessage `json:"params,omitempty"` Result json.RawMessage `json:"result,omitempty"` Error *struct { + Code int `json:"code"` Message string `json:"message"` } `json:"error,omitempty"` } @@ -114,7 +115,7 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j select { case m := <-answer: if m.Error != nil { - return nil, &refused{m.Error.Message} + return nil, &refused{message: m.Error.Message, code: m.Error.Code} } return m.Result, nil case <-time.After(timeout): @@ -126,10 +127,23 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j } // refused is the runtime's error answer, as opposed to no answer: what it refused, it did not do. -type refused struct{ message string } +type refused struct { + message string + code int +} func (r *refused) Error() string { return r.message } +// Is makes a compare-and-set the runtime refused because another writer was first ErrChanged. +func (r *refused) Is(target error) bool { return target == ErrChanged && r.code == CodeChanged } + +// ErrChanged is a compare-and-set that lost: the key was made or changed by another writer first (novox/hq +// ADR 0259). Told apart by the runtime's error code, never by its words. +var ErrChanged = errors.New("the key was written by another writer first") + +// CodeChanged is the error code the runtime answers a lost compare-and-set with. +const CodeChanged = -32010 + var ( current *channel currentMu sync.Mutex