mesh-mail-notifier: watch retries a subscribe refusal instead of exiting (issue 859, review round 1)
mesh/merge-gate head 31120ed58029 merged into trunk c9848f4fb85c — pass: builds mesh-mail-notifier → g14, shanks; no bus step; every machine composes with …
mesh/repo-check head 31120ed58029 merged into trunk c9848f4fb85c — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
mesh/merge-gate head 31120ed58029 merged into trunk c9848f4fb85c — pass: builds mesh-mail-notifier → g14, shanks; no bus step; every machine composes with …
mesh/repo-check head 31120ed58029 merged into trunk c9848f4fb85c — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
A subscribe the runtime refuses no longer calls os.Exit(1), which took mail-notifier.status and mail_notifier_health down with it: watch now retries with backoff (2s doubling to 1 minute), logging each refusal, as mesh-knowledge's hear does. Also: the run() test proves liveness by having the watcher answer a ping after run returns, instead of checking a channel only the test's own deferred cleanup could close; and the manifest test is renamed to say what it checks (no process resource at all).
This commit is contained in:
@@ -80,19 +80,28 @@ func run(n *Notifier, watcher func(*Notifier), in io.Reader, out io.Writer, tool
|
||||
// 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,
|
||||
// the lead's reading, review round 1 of issue 800).
|
||||
func watch(n *Notifier) {
|
||||
time.Sleep(500 * time.Millisecond) // Serve first: the bus is reached through it
|
||||
err := stdio.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
|
||||
func watch(n *Notifier) { watchWith(n, stdio.Subscribe, time.Sleep) }
|
||||
|
||||
// watchWith is watch with its subscribe and sleep factored out, so a test can make a subscribe fail without
|
||||
// the real bus and without the real wait. A subscribe the runtime refuses is retried, 2 seconds doubling to
|
||||
// at most a minute, logging each refusal — exactly as modules/mesh-knowledge's hear does — never fatal: a
|
||||
// refusal here must never take mail-notifier.status and mail_notifier_health down with it (novox/hq issue
|
||||
// 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)
|
||||
})
|
||||
if err != nil {
|
||||
logf("[mesh-mail-notifier] could not hear %s: %v", AddedPattern, err)
|
||||
os.Exit(1)
|
||||
logf("[mesh-mail-notifier] does not hear %s yet (%v); asking again in %s", AddedPattern, err, wait)
|
||||
}
|
||||
ticker := time.NewTicker(1 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
@@ -28,7 +28,7 @@ func readManifestResources(t *testing.T) manifestResources {
|
||||
return m
|
||||
}
|
||||
|
||||
func TestNoProcessNamedNotifyIsDeclared(t *testing.T) {
|
||||
func TestNoProcessResourceIsDeclared(t *testing.T) {
|
||||
m := readManifestResources(t)
|
||||
for _, r := range m.Resources {
|
||||
if r["type"] == "process" {
|
||||
|
||||
@@ -1,27 +1,33 @@
|
||||
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.
|
||||
// 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{})
|
||||
finished := make(chan struct{})
|
||||
stop := make(chan struct{})
|
||||
defer close(stop)
|
||||
ping := make(chan struct{})
|
||||
pong := make(chan struct{})
|
||||
|
||||
fakeWatcher := func(n *Notifier) {
|
||||
close(started)
|
||||
<-stop // the real watch never returns either: it loops until the process itself ends
|
||||
close(finished)
|
||||
for range ping {
|
||||
pong <- struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
n := &Notifier{Store: newFakeStore()}
|
||||
@@ -45,9 +51,64 @@ func TestRunStartsTheWatcherEvenWhenStdinIsAtEOF(t *testing.T) {
|
||||
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 <-finished:
|
||||
t.Fatal("the watcher must not stop just because stdio.Serve returned: it is the base invocation's own watcher now, not a separate `notify` process that exits at EOF")
|
||||
default:
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user