Merge pull request 'mesh-mail-notifier: the watcher runs in its runtime, not a process that exits at once (issue 859)' (#322) from fix/859-notifier-runs-in-its-runtime into main
This commit was merged in pull request #322.
This commit is contained in:
@@ -37,4 +37,4 @@ accounts it announces, in mesh-mail's trusted setting `grants` (`mail-store`'s s
|
|||||||
| tool | does |
|
| tool | does |
|
||||||
|---|---|
|
|---|---|
|
||||||
| `mail-notifier.status` | every account it is granted, when it last showed each one a notice, and any notice still failing to send (its count, last error, since when). Read-only |
|
| `mail-notifier.status` | every account it is granted, when it last showed each one a notice, and any notice still failing to send (its count, last error, since when). Read-only |
|
||||||
| `mail_notifier_health` | how it is: unhealthy once a notice has failed to show for more than ten minutes, healthy otherwise, even mid-retry. Not (yet) the node-engine's own check of `notify` — one process cannot be judged only by its own tool (ADR 0227 rule 8); `notify`'s manifest health stays `unit` until the lead decides how a one-process module gets this automatically. Read-only |
|
| `mail_notifier_health` | how it is: unhealthy once a notice has failed to show for more than ten minutes, healthy otherwise, even mid-retry. Not (yet) the node-engine's own check — this is one process, judged only by its own tool (ADR 0227 rule 8); a watching process of its own is issue 841, not yet decided. Read-only |
|
||||||
|
|||||||
@@ -1,13 +1,16 @@
|
|||||||
// mesh-mail-notifier: the default claimant of mail-notifier, built from mesh-mail's own event shape (novox/hq
|
// mesh-mail-notifier: the default claimant of mail-notifier, built from mesh-mail's own event shape (novox/hq
|
||||||
// ADR 0378 §9, issue 800 part B). A Go bundle the node's runtime launches, its own account (runs-as, ADR 0376
|
// ADR 0378 §9, issue 800 part B). A Go bundle the node's runtime launches, its own account (runs-as, ADR 0376
|
||||||
// §2): the base invocation serves mail-notifier.status and the module's own health tool
|
// §2): the one process serves mail-notifier.status and the module's own health tool (mail_notifier_health,
|
||||||
// (mail_notifier_health, review round 1 of issue 800); a separate process (`notify`) hears mesh-mail's added
|
// review round 1 of issue 800), and, beside stdio.Serve, hears mesh-mail's added events and shows them
|
||||||
// events and shows them. What this module says, it says on stderr; stdout is the MCP channel.
|
// (exactly as modules/mesh-knowledge's serve starts go hear(k) and go background(k) before stdio.Serve) — not
|
||||||
|
// a separate `notify` process, which systemd gave an empty stdin and so exited at once, every five seconds
|
||||||
|
// (novox/hq issue 859). What this module says, it says on stderr; stdout is the MCP channel.
|
||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
@@ -42,15 +45,6 @@ func main() {
|
|||||||
}
|
}
|
||||||
n.init()
|
n.init()
|
||||||
|
|
||||||
if len(os.Args) > 1 && os.Args[1] == "notify" {
|
|
||||||
go watch(n)
|
|
||||||
if err := stdio.Serve("", nil); err != nil {
|
|
||||||
logf("%v", err)
|
|
||||||
os.Exit(1)
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
tools := []stdio.Tool{
|
tools := []stdio.Tool{
|
||||||
{Name: StatusTool,
|
{Name: StatusTool,
|
||||||
Description: "The accounts mesh-mail-notifier announces on this machine, and when it last showed a " +
|
Description: "The accounts mesh-mail-notifier announces on this machine, and when it last showed a " +
|
||||||
@@ -67,28 +61,47 @@ func main() {
|
|||||||
return Judge(failures, time.Now()), nil
|
return Judge(failures, time.Now()), nil
|
||||||
}},
|
}},
|
||||||
}
|
}
|
||||||
if err := stdio.Serve("", tools); err != nil {
|
if err := run(n, watch, os.Stdin, os.Stdout, tools); err != nil {
|
||||||
logf("%v", err)
|
logf("%v", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// run is main's startup path, factored so a test can run it: the watcher started before stdio.Serve, exactly
|
||||||
|
// as mesh-knowledge's serve starts go hear(k) and go background(k) before stdio.Serve — so the watcher keeps
|
||||||
|
// running whatever stdio.Serve does with the stream it is given, including returning at once when stdin is
|
||||||
|
// already at EOF, as systemd gave the old separate `notify` process (novox/hq issue 859). The node's runtime
|
||||||
|
// launches this one process and keeps it alive.
|
||||||
|
func run(n *Notifier, watcher func(*Notifier), in io.Reader, out io.Writer, tools []stdio.Tool) error {
|
||||||
|
go watcher(n)
|
||||||
|
return stdio.ServeOn("", tools, in, out)
|
||||||
|
}
|
||||||
|
|
||||||
// watch hears mesh-mail's added events and, once a second, flushes what Notifier.Flush finds ready: an
|
// watch hears mesh-mail's added events and, once a second, flushes what Notifier.Flush finds ready: an
|
||||||
// account's first mails of its window at once, and its folded overflow once the window is over (ADR 0378 §9,
|
// account's first mails of its window at once, and its folded overflow once the window is over (ADR 0378 §9,
|
||||||
// the lead's reading, review round 1 of issue 800).
|
// the lead's reading, review round 1 of issue 800).
|
||||||
func watch(n *Notifier) {
|
func watch(n *Notifier) { watchWith(n, stdio.Subscribe, time.Sleep) }
|
||||||
time.Sleep(500 * time.Millisecond) // Serve first: the bus is reached through it
|
|
||||||
err := stdio.Subscribe(AddedPattern, func(e stdio.Envelope) error {
|
// watchWith is watch with its subscribe and sleep factored out, so a test can make a subscribe fail without
|
||||||
var body EventBody
|
// the real bus and without the real wait. A subscribe the runtime refuses is retried, 2 seconds doubling to
|
||||||
if err := json.Unmarshal(e.Body, &body); err != nil {
|
// at most a minute, logging each refusal — exactly as modules/mesh-knowledge's hear does — never fatal: a
|
||||||
logf("[mesh-mail-notifier] an added event could not be read; taken and dropped: %v", err)
|
// refusal here must never take mail-notifier.status and mail_notifier_health down with it (novox/hq issue
|
||||||
return nil
|
// 859, review round 1; the old os.Exit(1) did exactly that).
|
||||||
|
func watchWith(n *Notifier, subscribe func(string, func(stdio.Envelope) error) error, sleep func(time.Duration)) {
|
||||||
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||||
|
sleep(wait)
|
||||||
|
err := subscribe(AddedPattern, func(e stdio.Envelope) error {
|
||||||
|
var body EventBody
|
||||||
|
if err := json.Unmarshal(e.Body, &body); err != nil {
|
||||||
|
logf("[mesh-mail-notifier] an added event could not be read; taken and dropped: %v", err)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return n.Add(body)
|
||||||
|
})
|
||||||
|
if err == nil {
|
||||||
|
break
|
||||||
}
|
}
|
||||||
return n.Add(body)
|
logf("[mesh-mail-notifier] does not hear %s yet (%v); asking again in %s", AddedPattern, err, wait)
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
logf("[mesh-mail-notifier] could not hear %s: %v", AddedPattern, err)
|
|
||||||
os.Exit(1)
|
|
||||||
}
|
}
|
||||||
ticker := time.NewTicker(1 * time.Second)
|
ticker := time.NewTicker(1 * time.Second)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|||||||
@@ -0,0 +1,38 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// The manifest's own shape (novox/hq issue 859): the watch loop runs in the base invocation beside
|
||||||
|
// stdio.Serve, so no resource declares a separate `notify` process any more — the thing that exited at once,
|
||||||
|
// every five seconds, because a systemd-run process's stdin is empty and stdio.Serve returns at EOF.
|
||||||
|
|
||||||
|
type manifestResources struct {
|
||||||
|
Resources []map[string]any `json:"resources"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func readManifestResources(t *testing.T) manifestResources {
|
||||||
|
t.Helper()
|
||||||
|
raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var m manifestResources
|
||||||
|
if err := json.Unmarshal(raw, &m); err != nil {
|
||||||
|
t.Fatalf("module.json: %v", err)
|
||||||
|
}
|
||||||
|
return m
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNoProcessResourceIsDeclared(t *testing.T) {
|
||||||
|
m := readManifestResources(t)
|
||||||
|
for _, r := range m.Resources {
|
||||||
|
if r["type"] == "process" {
|
||||||
|
t.Fatalf("module.json still declares a process resource (%v): the watcher runs in the base invocation now, not a separate process (novox/hq issue 859)", r)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,114 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"io"
|
||||||
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestRunStartsTheWatcherEvenWhenStdinIsAtEOF proves the watcher runs beside stdio.Serve in the one process the
|
||||||
|
// node's runtime launches (novox/hq ADR 0378 §9, as mesh-knowledge's serve starts go hear(k) and go
|
||||||
|
// background(k) before stdio.Serve), not only in a separate process something keeps asking over stdin. run is
|
||||||
|
// given stdin already at EOF, exactly what systemd gave the old separate `notify` process: it exited at once,
|
||||||
|
// every five seconds (novox/hq issue 859). The watcher must still be running once run has returned — proved by
|
||||||
|
// making it actually answer after run returns, not by the absence of a signal that nothing but the test's own
|
||||||
|
// cleanup could ever raise.
|
||||||
|
func TestRunStartsTheWatcherEvenWhenStdinIsAtEOF(t *testing.T) {
|
||||||
|
started := make(chan struct{})
|
||||||
|
ping := make(chan struct{})
|
||||||
|
pong := make(chan struct{})
|
||||||
|
|
||||||
|
fakeWatcher := func(n *Notifier) {
|
||||||
|
close(started)
|
||||||
|
for range ping {
|
||||||
|
pong <- struct{}{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
n := &Notifier{Store: newFakeStore()}
|
||||||
|
n.init()
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- run(n, fakeWatcher, strings.NewReader(""), io.Discard, nil) }()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-started:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("the watcher was not started before stdio.Serve returned")
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("run over stdin at EOF: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("run did not return once stdin reached EOF, as systemd gives a process nobody talks to")
|
||||||
|
}
|
||||||
|
|
||||||
|
// run has returned (stdio.Serve hit EOF at once); the watcher must still be live and answering, not merely
|
||||||
|
// not yet torn down. A watcher that stopped when run returned would never take this ping.
|
||||||
|
select {
|
||||||
|
case ping <- struct{}{}:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("the watcher is no longer running once run returned: it must not stop just because stdio.Serve did")
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-pong:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("the watcher did not answer once run returned")
|
||||||
|
}
|
||||||
|
close(ping)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestWatchRetriesASubscribeRefusalWithBackoffAndNeverExits proves watch no longer exits the process on a
|
||||||
|
// subscribe the runtime refuses — the old os.Exit(1) took mail-notifier.status and mail_notifier_health down
|
||||||
|
// with it — but retries with backoff and logs each refusal, as modules/mesh-knowledge's hear does (novox/hq
|
||||||
|
// issue 859, review round 1). A subscribe that fails twice then succeeds must leave the process serving (run
|
||||||
|
// still returns normally over stdin at EOF, never os.Exit) and the watcher subscribed (its third, successful
|
||||||
|
// call).
|
||||||
|
func TestWatchRetriesASubscribeRefusalWithBackoffAndNeverExits(t *testing.T) {
|
||||||
|
var calls int32
|
||||||
|
subscribed := make(chan struct{})
|
||||||
|
fakeSubscribe := func(pattern string, handle func(stdio.Envelope) error) error {
|
||||||
|
if atomic.AddInt32(&calls, 1) <= 2 {
|
||||||
|
return errors.New("the runtime refused this subscribe")
|
||||||
|
}
|
||||||
|
close(subscribed)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
noSleep := func(time.Duration) {}
|
||||||
|
|
||||||
|
n := &Notifier{Store: newFakeStore()}
|
||||||
|
n.init()
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
done <- run(n, func(n *Notifier) { watchWith(n, fakeSubscribe, noSleep) },
|
||||||
|
strings.NewReader(""), io.Discard, nil)
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("run over stdin at EOF: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("run did not return once stdin reached EOF: a subscribe refusal must not hold up or kill the process")
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-subscribed:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("the watcher never subscribed after its first two refusals")
|
||||||
|
}
|
||||||
|
|
||||||
|
if got := atomic.LoadInt32(&calls); got != 3 {
|
||||||
|
t.Fatalf("subscribe was called %d time(s), want 3 (two refusals, then the one that succeeded)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -47,19 +47,6 @@
|
|||||||
"mode": "0700",
|
"mode": "0700",
|
||||||
"owner": "mesh-mail-notifier",
|
"owner": "mesh-mail-notifier",
|
||||||
"place": "."
|
"place": "."
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "notify",
|
|
||||||
"type": "process",
|
|
||||||
"name": "mesh-mail-notifier-notify",
|
|
||||||
"artifact": "tools",
|
|
||||||
"run": [
|
|
||||||
"./mesh-mail-notifier",
|
|
||||||
"notify"
|
|
||||||
],
|
|
||||||
"health": {
|
|
||||||
"kind": "unit"
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
"build": {
|
"build": {
|
||||||
|
|||||||
Reference in New Issue
Block a user