diff --git a/backend/internal/readerrpc/config.go b/backend/internal/readerrpc/config.go
index c3e76815..996da8bb 100644
--- a/backend/internal/readerrpc/config.go
+++ b/backend/internal/readerrpc/config.go
@@ -106,7 +106,7 @@ type Capabilities struct {
Daemon *DaemonInfo `json:"daemon,omitempty"`
}
-// Status is the response to Reader.GetStatus and the status-topic payload.
+// Status is the response to Reader.GetStatus. The status topic carries StatusDoc.
type Status struct {
Online bool `json:"online"`
Reading bool `json:"reading"`
diff --git a/backend/internal/readerrpc/rpc.go b/backend/internal/readerrpc/rpc.go
index b1e128b9..b36920b1 100644
--- a/backend/internal/readerrpc/rpc.go
+++ b/backend/internal/readerrpc/rpc.go
@@ -22,14 +22,34 @@ const (
MethodGetOperProfile = "Reader.GetOperProfile"
MethodSetOperProfile = "Reader.SetOperProfile"
MethodGetStatus = "Reader.GetStatus"
+ MethodGpoSet = "Gpo.Set"
+
+ // MethodReboot and MethodShutdown cleanly reboot or power off the reader
+ // host (sync, then systemd). The daemon replies first and acts after a short
+ // delay, announcing the planned outage on the retained status first.
+ // Neither takes required params; PowerParams is optional.
+ MethodReboot = "Reader.Reboot"
+ MethodShutdown = "Reader.Shutdown"
// Reserved for future use.
MethodScanStart = "Scan.Start"
MethodScanStop = "Scan.Stop"
- MethodGpoSet = "Gpo.Set"
- MethodReboot = "Reader.Reboot"
)
+// PowerParams are the optional params of Reader.Reboot and Reader.Shutdown.
+type PowerParams struct {
+ // RequestedBy names the person or system asking, for the reader's log. The
+ // request's Src is logged as well.
+ RequestedBy string `json:"requested_by,omitempty"`
+}
+
+// PowerResult acknowledges Reader.Reboot or Reader.Shutdown. The action has not
+// happened yet: it starts InMs after the reply.
+type PowerResult struct {
+ Action string `json:"action"` // StatusReasonReboot or StatusReasonShutdown
+ InMs int `json:"in_ms"`
+}
+
// Standard JSON-RPC 2.0 error codes.
const (
CodeParseError = -32700
diff --git a/mqtt-rpc/README.md b/mqtt-rpc/README.md
index 83263af8..ef96c33d 100644
--- a/mqtt-rpc/README.md
+++ b/mqtt-rpc/README.md
@@ -55,10 +55,11 @@ never evicts the reader's own reads connection.
**Capability-flagged management** (declared per adapter in `supports[]` / `unsupported[]`):
- `Gpo.Set(params={port, on, pulse_ms?})` → `{port, on, pulsed}` — supported by the CS463 adapter. See "Alarm outputs".
-- `Scan.Start`, `Scan.Stop`, `Reader.Reboot` — reserved; `unsupported` on the CS463.
+- `Scan.Start`, `Scan.Stop` — reserved; `unsupported` on the CS463.
**Daemon** (acts on mqtt-rpcd itself; never touches the vendor `/API`):
- `Daemon.Update(params={version, url, sha256, signature})` → `{accepted, from, to}` once verified; the install follows the reply. Refusals are error `-33002`. See "Remote update".
+- `Reader.Reboot`, `Reader.Shutdown(params={requested_by?})` → `{action: "reboot"|"shutdown", in_ms}`; the clean reboot or power-off follows the reply. Listed in `supports[]` when the daemon runs on a host it can reboot. See "Reboot and shutdown".
Rules: neutral config (no vendor terms like "operation profile" leak into the
wire); a new tunable is a field, not a method; `applied` is explicit; the contract
@@ -262,6 +263,32 @@ reconcile re-arms it.
retry, but takes no recovery action. Use it on a bench that has to hold a wedge
for diagnosis.
+If systemd refuses the reboot, the reader is still up: the daemon republishes
+the status online and tries again after the next window.
+
+### Reboot and shutdown (`Reader.Reboot`, `Reader.Shutdown`)
+A clean reboot or power-off an operator asks for over the broker, for a reader
+with no SSH. A reboot is the known fix for the vendor-app wedge; a shutdown
+before planned power work avoids the unclean power cut that causes it.
+
+```sh
+MQTT_URL=… mqtt-rpcctl reboot -base trakrf.id/cs463-212 -yes
+MQTT_URL=… mqtt-rpcctl shutdown -base trakrf.id/cs463-212 -yes # back only after an on-site power-cycle
+```
+
+1. The daemon logs the request (`requested_by`, the reply topic, `asked_at`) and
+ replies `{action, in_ms: 2000}`.
+2. 2 s later it publishes the retained status offline with `reason: "reboot"`
+ or `"shutdown"`, then runs `sync` and `systemctl reboot` or
+ `systemctl poweroff`. Never a power reset. systemd then stops the daemon,
+ whose stop drives every output off and keeps the reason on the status.
+3. A second request while one is under way, or while an automatic recovery
+ reboot is announced, is refused (`-32603`, "already going down").
+4. If systemd refuses, the status goes back online and a retry is accepted.
+
+After a reboot the daemon takes 2–7 min to reach the broker again (boot-time
+entropy for TLS).
+
### Status document (retained `/status`)
```jsonc
{"online": true, "at": "2026-11-08T17:40:00Z", "version": "v1.5.0-330-g…",
@@ -271,7 +298,8 @@ for diagnosis.
"auto_reboots": 1, "auto_reboots_exhausted": false,
"golden_config": "pending", // "applied" | "pending" | "off"
"cloudserver_fault": "cleanSession is false; golden is true"}} // absent when clean
-// before an automatic reboot, and after it via Stop:
+// before a planned outage, and after it via Stop. reason is "auto-recovery"
+// (the daemon's own), or "reboot" / "shutdown" (Reader.Reboot / Reader.Shutdown):
{"online": false, "reason": "auto-recovery", "version": "…", "health": {…}}
// MQTT Last Will (the daemon died without Stop):
{"online": false}
diff --git a/mqtt-rpc/cmd/mqtt-rpcctl/main.go b/mqtt-rpc/cmd/mqtt-rpcctl/main.go
index d90fea52..01733527 100644
--- a/mqtt-rpc/cmd/mqtt-rpcctl/main.go
+++ b/mqtt-rpc/cmd/mqtt-rpcctl/main.go
@@ -7,9 +7,15 @@
// mqtt-rpcctl call -base TOPIC -method M [-params JSON] [-ca FILE]
// mqtt-rpcctl update -base TOPIC -manifest FILE|URL [-ca FILE]
// mqtt-rpcctl status -base TOPIC [-watch] [-ca FILE]
+// mqtt-rpcctl reboot -base TOPIC -yes [-by WHO] [-ca FILE]
+// mqtt-rpcctl shutdown -base TOPIC -yes [-by WHO] [-ca FILE]
+//
+// reboot and shutdown are clean (sync, then systemd), and both take the reader
+// offline: a shutdown needs someone on site to power-cycle it back. -by goes to
+// the reader's log and defaults to $USER@hostname.
//
// sign reads the key from -key or, if absent, from $MQTT_RPCD_SIGNING_KEY.
-// call/update/status read the broker URL (with credentials) from $MQTT_URL so they never
+// call/update/status/reboot/shutdown read the broker URL (with credentials) from $MQTT_URL so they never
// appear on a command line. -base is the reader's base topic, e.g.
// trakrf.id/cs463-212.
package main
@@ -54,6 +60,8 @@ func main() {
err = cmdUpdate(os.Args[2:])
case "status":
err = cmdStatus(os.Args[2:])
+ case "reboot", "shutdown":
+ err = cmdPower(os.Args[1], os.Args[2:])
default:
usage()
}
@@ -64,7 +72,7 @@ func main() {
}
func usage() {
- fmt.Fprintln(os.Stderr, "usage: mqtt-rpcctl keygen|sign|call|update|status [flags] (see -h on each)")
+ fmt.Fprintln(os.Stderr, "usage: mqtt-rpcctl keygen|sign|call|update|status|reboot|shutdown [flags] (see -h on each)")
os.Exit(2)
}
diff --git a/mqtt-rpc/cmd/mqtt-rpcctl/power.go b/mqtt-rpc/cmd/mqtt-rpcctl/power.go
new file mode 100644
index 00000000..b76ec338
--- /dev/null
+++ b/mqtt-rpc/cmd/mqtt-rpcctl/power.go
@@ -0,0 +1,54 @@
+package main
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+
+ "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc"
+)
+
+// cmdPower handles `reboot` and `shutdown`: a clean reboot or power-off of the
+// reader host. Both take the reader offline, so both need -yes.
+func cmdPower(action string, args []string) error {
+ fs, base, ca := rpcFlags(action)
+ yes := fs.Bool("yes", false, "confirm: this takes the reader offline")
+ by := fs.String("by", defaultRequestedBy(), "who is asking, for the reader's log")
+ _ = fs.Parse(args)
+ if *base == "" {
+ return errors.New("-base is required")
+ }
+ method, params, err := powerRequest(action, *yes, *by)
+ if err != nil {
+ return err
+ }
+ return rpc(*base, *ca, method, params)
+}
+
+// powerRequest builds the Reader.Reboot or Reader.Shutdown request, refusing
+// without confirmation.
+func powerRequest(action string, yes bool, by string) (string, json.RawMessage, error) {
+ var method, cost string
+ switch action {
+ case "reboot":
+ method = readerrpc.MethodReboot
+ cost = "the reader stops reading and is off the broker for several minutes (2-7 after boot)"
+ case "shutdown":
+ method = readerrpc.MethodShutdown
+ cost = "the reader powers off and comes back only when someone on site power-cycles it"
+ default:
+ return "", nil, fmt.Errorf("unknown power action %q", action)
+ }
+ if !yes {
+ return "", nil, fmt.Errorf("%s: %s; pass -yes to go ahead", action, cost)
+ }
+ params, err := json.Marshal(readerrpc.PowerParams{RequestedBy: by})
+ return method, params, err
+}
+
+// defaultRequestedBy is $USER@hostname, or whichever half is known.
+func defaultRequestedBy() string {
+ host, _ := os.Hostname()
+ return os.Getenv("USER") + "@" + host
+}
diff --git a/mqtt-rpc/cmd/mqtt-rpcctl/power_test.go b/mqtt-rpc/cmd/mqtt-rpcctl/power_test.go
new file mode 100644
index 00000000..ae240edd
--- /dev/null
+++ b/mqtt-rpc/cmd/mqtt-rpcctl/power_test.go
@@ -0,0 +1,55 @@
+package main
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+
+ "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc"
+)
+
+func TestPowerRequest(t *testing.T) {
+ cases := []struct {
+ name, action string
+ yes bool
+ wantMethod string
+ wantErr string // substring; empty means no error
+ }{
+ {"reboot needs -yes", "reboot", false, "", "-yes"},
+ {"shutdown needs -yes and says what it costs", "shutdown", false, "", "power-cycle"},
+ {"reboot", "reboot", true, readerrpc.MethodReboot, ""},
+ {"shutdown", "shutdown", true, readerrpc.MethodShutdown, ""},
+ {"unknown action", "halt", true, "", "unknown"},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ method, params, err := powerRequest(c.action, c.yes, "mike@bench")
+ if c.wantErr != "" {
+ if err == nil || !strings.Contains(err.Error(), c.wantErr) {
+ t.Fatalf("err = %v, want one containing %q", err, c.wantErr)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if method != c.wantMethod {
+ t.Errorf("method = %q, want %q", method, c.wantMethod)
+ }
+ var p readerrpc.PowerParams
+ if err := json.Unmarshal(params, &p); err != nil {
+ t.Fatal(err)
+ }
+ if p.RequestedBy != "mike@bench" {
+ t.Errorf("requested_by = %q, want mike@bench", p.RequestedBy)
+ }
+ })
+ }
+}
+
+func TestDefaultRequestedBy(t *testing.T) {
+ t.Setenv("USER", "mike")
+ if got := defaultRequestedBy(); !strings.HasPrefix(got, "mike@") || got == "mike@" {
+ t.Errorf("defaultRequestedBy = %q, want mike@", got)
+ }
+}
diff --git a/mqtt-rpc/cmd/mqtt-rpcctl/status_test.go b/mqtt-rpc/cmd/mqtt-rpcctl/status_test.go
index 46b33c7c..e0460c20 100644
--- a/mqtt-rpc/cmd/mqtt-rpcctl/status_test.go
+++ b/mqtt-rpc/cmd/mqtt-rpcctl/status_test.go
@@ -14,6 +14,10 @@ func TestSummarizeStatus(t *testing.T) {
`online v1 api=healthy golden=applied cloudserver_fault="cleanSession is false; golden is true"`},
{"planned offline", `{"online":false,"reason":"auto-recovery","version":"v1"}`,
"offline (auto-recovery) v1"},
+ {"operator reboot", `{"online":false,"reason":"reboot","version":"v1"}`,
+ "offline (reboot) v1"},
+ {"operator shutdown", `{"online":false,"reason":"shutdown","version":"v1"}`,
+ "offline (shutdown) v1"},
{"last will", `{"online":false}`, "offline"},
{"older daemon without health", `{"online":true,"version":"v0"}`, "online v0"},
{"not json", `garbage`, "unparseable: garbage"},
diff --git a/mqtt-rpc/cmd/mqtt-rpcd/main.go b/mqtt-rpc/cmd/mqtt-rpcd/main.go
index fbf01209..11371987 100644
--- a/mqtt-rpc/cmd/mqtt-rpcd/main.go
+++ b/mqtt-rpc/cmd/mqtt-rpcd/main.go
@@ -100,8 +100,12 @@ func main() {
}
}
+ // One host for the operator's Reader.Reboot/Shutdown and the monitor's
+ // automatic recovery.
+ host := readerd.NewSystemdHost(&logger)
d := readerd.New(cfg, adapter, &logger)
d.SetUpdater(upd)
+ d.SetHost(host)
if err := d.Start(); err != nil {
logger.Fatal().Err(err).Msg("start daemon")
}
@@ -111,12 +115,13 @@ func main() {
policy.AutoRecover = cfg.AutoRecover
policy.WedgeAfter = cfg.WedgeAfter
monitor := readerd.NewMonitor(policy, readerd.MonitorDeps{
- Probe: adapter.ProbeHealth,
- Reconcile: reconcile,
- Host: readerd.NewSystemdHost(&logger),
- Store: readerd.RebootStore{Path: readerd.DefaultRebootStorePath},
- Publish: d.SetHealth,
- GoingDown: d.GoingDown,
+ Probe: adapter.ProbeHealth,
+ Reconcile: reconcile,
+ Host: host,
+ Store: readerd.RebootStore{Path: readerd.DefaultRebootStorePath},
+ Publish: d.SetHealth,
+ GoingDown: d.GoingDown,
+ AbortGoingDown: d.AbortGoingDown,
CloudServer: func() string {
return d.CloudServerFault(cfg.CloudServer, time.Now())
},
diff --git a/mqtt-rpc/internal/readerd/cs463/adapter.go b/mqtt-rpc/internal/readerd/cs463/adapter.go
index 6198aaf0..f02d5ac2 100644
--- a/mqtt-rpc/internal/readerd/cs463/adapter.go
+++ b/mqtt-rpc/internal/readerd/cs463/adapter.go
@@ -234,10 +234,11 @@ func (a *Adapter) GetCapabilities(ctx context.Context) (readerrpc.Capabilities,
readerrpc.MethodGetStatus,
readerrpc.MethodGpoSet,
},
+ // Reader.Reboot and Reader.Shutdown run on the host, not the reader API;
+ // the daemon advertises them (readerd.Daemon.SetHost).
Unsupported: []string{
readerrpc.MethodScanStart,
readerrpc.MethodScanStop,
- readerrpc.MethodReboot,
},
}, nil
}
diff --git a/mqtt-rpc/internal/readerd/cs463/adapter_test.go b/mqtt-rpc/internal/readerd/cs463/adapter_test.go
index 3929e9b8..ff272d4f 100644
--- a/mqtt-rpc/internal/readerd/cs463/adapter_test.go
+++ b/mqtt-rpc/internal/readerd/cs463/adapter_test.go
@@ -214,10 +214,11 @@ func TestGetCapabilities(t *testing.T) {
if !reflect.DeepEqual(caps.Supports, wantSupports) {
t.Errorf("supports = %v, want %v", caps.Supports, wantSupports)
}
+ // Reader.Reboot and Reader.Shutdown are the daemon's, not the adapter's: it
+ // advertises them itself when it has a host to run them on.
wantUnsupported := []string{
readerrpc.MethodScanStart,
readerrpc.MethodScanStop,
- readerrpc.MethodReboot,
}
if !reflect.DeepEqual(caps.Unsupported, wantUnsupported) {
t.Errorf("unsupported = %v, want %v", caps.Unsupported, wantUnsupported)
diff --git a/mqtt-rpc/internal/readerd/daemon.go b/mqtt-rpc/internal/readerd/daemon.go
index 978490d4..a52d4b58 100644
--- a/mqtt-rpc/internal/readerd/daemon.go
+++ b/mqtt-rpc/internal/readerd/daemon.go
@@ -52,9 +52,16 @@ type Daemon struct {
// publishRetained is the retained-status seam, set by Start (status.go).
publishRetained func(topic string, payload []byte) error
- statusMu sync.Mutex
- health *readerrpc.Health // latest from the health monitor
- reason string // set by GoingDown before a planned reboot
+ // host runs Reader.Reboot and Reader.Shutdown (SetHost); nil disables them.
+ // after schedules the delayed power action; a seam for tests.
+ host Host
+ powerDelay time.Duration
+ after func(time.Duration, func())
+
+ statusMu sync.Mutex
+ health *readerrpc.Health // latest from the health monitor
+ reason string // set by GoingDown before a planned reboot or power-off
+ powerPending string // a Reader.Reboot/Shutdown accepted and not yet acted on
anchor anchorCheck // the last broker handshake, judged as CSL would
}
@@ -83,6 +90,9 @@ func New(cfg Config, adapter Adapter, log *zerolog.Logger) *Daemon {
started: time.Now(),
commitAfter: DefaultCommitAfter,
+
+ powerDelay: DefaultPowerDelay,
+ after: func(wait time.Duration, f func()) { time.AfterFunc(wait, f) },
}
}
@@ -311,11 +321,17 @@ func (d *Daemon) dispatch(ctx context.Context, req readerrpc.Request) readerrpc.
if err != nil {
return readerrpc.NewError(req, readerrpc.CodeInternal, err.Error())
}
- return d.result(req, d.withDaemonInfo(caps))
+ return d.result(req, d.withPowerMethods(d.withDaemonInfo(caps)))
case readerrpc.MethodDaemonUpdate:
return d.daemonUpdate(req)
+ case readerrpc.MethodReboot:
+ return d.power(req, readerrpc.StatusReasonReboot)
+
+ case readerrpc.MethodShutdown:
+ return d.power(req, readerrpc.StatusReasonShutdown)
+
case readerrpc.MethodGetOperProfile:
var p readerrpc.OperProfileParams
if len(req.Params) > 0 {
diff --git a/mqtt-rpc/internal/readerd/health.go b/mqtt-rpc/internal/readerd/health.go
index 87cd2874..e7b1c917 100644
--- a/mqtt-rpc/internal/readerd/health.go
+++ b/mqtt-rpc/internal/readerd/health.go
@@ -70,6 +70,9 @@ type MonitorDeps struct {
Publish func(readerrpc.Health)
// GoingDown is called just before an automatic reboot, with the reason.
GoingDown func(reason string)
+ // AbortGoingDown takes GoingDown back when the reboot was refused, so a
+ // reader that stayed up is not announced as offline.
+ AbortGoingDown func()
// CloudServer returns the golden CloudServer fault ("" when clean), re-run
// every tick so an operator's fix, or a firmware update reverting one,
// shows without a daemon restart. nil checks nothing.
@@ -298,6 +301,9 @@ func (m *Monitor) reboot(now time.Time, h *readerrpc.Health) {
}
if err := m.deps.Host.Reboot(); err != nil {
m.log.Error().Err(err).Msg("auto-recovery: reboot failed; retrying after the next window")
+ if m.deps.AbortGoingDown != nil {
+ m.deps.AbortGoingDown()
+ }
}
}
diff --git a/mqtt-rpc/internal/readerd/health_test.go b/mqtt-rpc/internal/readerd/health_test.go
index 6469c3a7..affec8ff 100644
--- a/mqtt-rpc/internal/readerd/health_test.go
+++ b/mqtt-rpc/internal/readerd/health_test.go
@@ -31,6 +31,7 @@ type rig struct {
events []string
published []readerrpc.Health
store RebootStore
+ rebootErr error // what the host's Reboot returns
}
type rigHost struct{ r *rig }
@@ -41,6 +42,10 @@ func (h rigHost) RestartReaderApp(context.Context) error {
}
func (h rigHost) Reboot() error {
h.r.events = append(h.r.events, "reboot")
+ return h.r.rebootErr
+}
+func (h rigHost) Poweroff() error {
+ h.r.events = append(h.r.events, "poweroff")
return nil
}
@@ -64,9 +69,10 @@ func newRig(t *testing.T, p HealthPolicy, reconcile bool) *rig {
Publish: func(h readerrpc.Health) {
r.published = append(r.published, h)
},
- GoingDown: func(reason string) { r.events = append(r.events, "going-down:"+reason) },
- Log: &log,
- Now: func() time.Time { return r.now },
+ GoingDown: func(reason string) { r.events = append(r.events, "going-down:"+reason) },
+ AbortGoingDown: func() { r.events = append(r.events, "abort-going-down") },
+ Log: &log,
+ Now: func() time.Time { return r.now },
}
if reconcile {
deps.Reconcile = func(context.Context) error {
@@ -212,6 +218,26 @@ func TestMonitor_WedgeLadder(t *testing.T) {
}
}
+// A reboot systemd refused leaves the reader up. The monitor announced it as
+// going down first, so it must take that back, or the retained status says
+// offline for as long as the daemon keeps running.
+func TestMonitor_FailedRebootTakesBackTheOfflineStatus(t *testing.T) {
+ p := DefaultHealthPolicy()
+ r := newRig(t, p, true)
+ r.rebootErr = errors.New("systemctl reboot: exit status 1")
+ r.run(H, time.Minute)
+ r.run(D, p.WedgeAfter+p.Interval) // app restart
+ r.run(D, p.WedgeAfter) // reboot, refused
+
+ if r.count("reboot") != 1 {
+ t.Fatalf("events %q, want one reboot attempt", r.events)
+ }
+ tail := strings.Join(r.events[len(r.events)-3:], ",")
+ if tail != "going-down:auto-recovery,reboot,abort-going-down" {
+ t.Fatalf("failed reboot sequence = %q, want the offline status taken back", tail)
+ }
+}
+
func TestMonitor_RecoveryAfterAppRestartClearsWedge(t *testing.T) {
p := DefaultHealthPolicy()
r := newRig(t, p, true)
diff --git a/mqtt-rpc/internal/readerd/host.go b/mqtt-rpc/internal/readerd/host.go
index 64cac407..c93fffaa 100644
--- a/mqtt-rpc/internal/readerd/host.go
+++ b/mqtt-rpc/internal/readerd/host.go
@@ -22,6 +22,9 @@ type Host interface {
RestartReaderApp(ctx context.Context) error
// Reboot asks systemd for a clean reboot. It returns once the job is queued.
Reboot() error
+ // Poweroff asks systemd for a clean power-off. It returns once the job is
+ // queued. Bringing the reader back needs a power-cycle on site.
+ Poweroff() error
}
const (
@@ -167,3 +170,10 @@ func (h *SystemdHost) Reboot() error {
h.sync()
return h.run(context.Background(), "systemctl", "reboot")
}
+
+// Poweroff flushes the filesystems and asks systemd for a clean power-off: the
+// orderly shutdown that avoids the vendor-app wedge an unclean power cut causes.
+func (h *SystemdHost) Poweroff() error {
+ h.sync()
+ return h.run(context.Background(), "systemctl", "poweroff")
+}
diff --git a/mqtt-rpc/internal/readerd/host_test.go b/mqtt-rpc/internal/readerd/host_test.go
index f29a340e..fe3b5900 100644
--- a/mqtt-rpc/internal/readerd/host_test.go
+++ b/mqtt-rpc/internal/readerd/host_test.go
@@ -121,6 +121,17 @@ func TestReboot_SyncsFirst(t *testing.T) {
}
}
+func TestPoweroff_SyncsFirst(t *testing.T) {
+ fs := &fakeSystem{}
+ h := newFakeSystemdHost(t, fs)
+ if err := h.Poweroff(); err != nil {
+ t.Fatal(err)
+ }
+ if strings.Join(fs.calls, "|") != "sync|systemctl poweroff" {
+ t.Fatalf("calls = %q, want sync then systemctl poweroff", fs.calls)
+ }
+}
+
func TestReadCgroupProcs(t *testing.T) {
path := filepath.Join(t.TempDir(), "cgroup.procs")
if err := os.WriteFile(path, []byte("492\n879\n\n"), 0o644); err != nil {
diff --git a/mqtt-rpc/internal/readerd/power.go b/mqtt-rpc/internal/readerd/power.go
new file mode 100644
index 00000000..e47c80b3
--- /dev/null
+++ b/mqtt-rpc/internal/readerd/power.go
@@ -0,0 +1,101 @@
+package readerd
+
+import (
+ "encoding/json"
+ "time"
+
+ "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc"
+)
+
+// DefaultPowerDelay is how long Reader.Reboot and Reader.Shutdown wait after
+// replying before they act, so the caller gets the acknowledgement before the
+// daemon's broker connection goes away.
+const DefaultPowerDelay = 2 * time.Second
+
+// SetHost enables Reader.Reboot and Reader.Shutdown. Call before Start.
+func (d *Daemon) SetHost(h Host) { d.host = h }
+
+// power handles Reader.Reboot and Reader.Shutdown: validate, refuse a second
+// request while one is under way, log who asked, reply, and act after
+// powerDelay. action is StatusReasonReboot or StatusReasonShutdown.
+func (d *Daemon) power(req readerrpc.Request, action string) readerrpc.Response {
+ if d.host == nil {
+ return readerrpc.NewError(req, readerrpc.CodeMethodNotFound, "unsupported method: "+req.Method)
+ }
+ var p readerrpc.PowerParams
+ if len(req.Params) > 0 {
+ if err := json.Unmarshal(req.Params, &p); err != nil {
+ return readerrpc.NewError(req, readerrpc.CodeInvalidParams, "invalid "+req.Method+" params: "+err.Error())
+ }
+ }
+
+ d.statusMu.Lock()
+ busy := d.reason // an automatic recovery reboot already announced
+ if busy == "" {
+ busy = d.powerPending
+ }
+ if busy == "" {
+ d.powerPending = action
+ }
+ d.statusMu.Unlock()
+ if busy != "" {
+ return readerrpc.NewError(req, readerrpc.CodeInternal, "already going down: "+busy)
+ }
+
+ delay := d.powerDelay
+ d.log.Warn().
+ Str("method", req.Method).
+ Str("requested_by", p.RequestedBy).
+ Str("src", req.Src).
+ Int("id", req.ID).
+ Time("asked_at", d.now().UTC()).
+ Dur("in", delay).
+ Msg("power action requested over rpc")
+ d.after(delay, func() { d.powerNow(action) })
+ return d.result(req, readerrpc.PowerResult{Action: action, InMs: int(delay / time.Millisecond)})
+}
+
+// powerNow announces the planned outage on the retained status, then asks
+// systemd. If systemd refuses, the reader is still up, so it must not stay
+// announced as offline.
+func (d *Daemon) powerNow(action string) {
+ d.GoingDown(action)
+ run := d.host.Reboot
+ if action == readerrpc.StatusReasonShutdown {
+ run = d.host.Poweroff
+ }
+ if err := run(); err != nil {
+ d.log.Error().Err(err).Str("action", action).Msg("power action failed; the reader stays up")
+ d.AbortGoingDown()
+ return
+ }
+ d.log.Warn().Str("action", action).Msg("power action queued with systemd")
+}
+
+// AbortGoingDown undoes GoingDown after a reboot or power-off that did not
+// happen: it clears the reason and republishes the reader as online.
+func (d *Daemon) AbortGoingDown() {
+ d.statusMu.Lock()
+ d.reason = ""
+ d.powerPending = ""
+ d.statusMu.Unlock()
+ d.publishStatus(d.onlinePayload())
+}
+
+// withPowerMethods advertises Reader.Reboot and Reader.Shutdown when the daemon
+// has a host to run them on.
+func (d *Daemon) withPowerMethods(caps readerrpc.Capabilities) readerrpc.Capabilities {
+ if d.host == nil {
+ return caps
+ }
+ power := []string{readerrpc.MethodReboot, readerrpc.MethodShutdown}
+ caps.Supports = append(append([]string(nil), caps.Supports...), power...)
+ var unsupported []string
+ for _, m := range caps.Unsupported {
+ if m != readerrpc.MethodReboot && m != readerrpc.MethodShutdown {
+ unsupported = append(unsupported, m)
+ }
+ }
+ caps.Unsupported = unsupported
+ return caps
+}
diff --git a/mqtt-rpc/internal/readerd/power_test.go b/mqtt-rpc/internal/readerd/power_test.go
new file mode 100644
index 00000000..18ccf894
--- /dev/null
+++ b/mqtt-rpc/internal/readerd/power_test.go
@@ -0,0 +1,218 @@
+package readerd
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/rs/zerolog"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc"
+)
+
+// powerRig is a daemon with a fake host, a captured retained status, and a
+// held timer: the delayed action runs only when the test fires it.
+type powerRig struct {
+ d *Daemon
+ events []string // retained status docs and host calls, in order
+ status []readerrpc.StatusDoc
+ pending []func()
+ delays []time.Duration
+ hostErr error
+ logs bytes.Buffer
+}
+
+type powerHost struct{ r *powerRig }
+
+func (h powerHost) RestartReaderApp(context.Context) error { return nil }
+func (h powerHost) Reboot() error {
+ h.r.events = append(h.r.events, "host reboot")
+ return h.r.hostErr
+}
+func (h powerHost) Poweroff() error {
+ h.r.events = append(h.r.events, "host poweroff")
+ return h.r.hostErr
+}
+
+func newPowerRig(t *testing.T, withHost bool) *powerRig {
+ t.Helper()
+ r := &powerRig{}
+ log := zerolog.New(&r.logs)
+ r.d = New(Config{Broker: BrokerConfig{BaseTopic: "trakrf.id/cs463-212"}}, &fakeAdapter{}, &log)
+ if withHost {
+ r.d.SetHost(powerHost{r})
+ }
+ r.d.after = func(d time.Duration, f func()) {
+ r.delays = append(r.delays, d)
+ r.pending = append(r.pending, f)
+ }
+ r.d.publishRetained = func(_ string, b []byte) error {
+ var doc readerrpc.StatusDoc
+ require.NoError(t, json.Unmarshal(b, &doc))
+ r.status = append(r.status, doc)
+ if doc.Online {
+ r.events = append(r.events, "status online")
+ } else {
+ r.events = append(r.events, "status offline "+doc.Reason)
+ }
+ return nil
+ }
+ return r
+}
+
+func (r *powerRig) call(t *testing.T, method, params string) readerrpc.Response {
+ t.Helper()
+ req := readerrpc.Request{ID: 9, Src: "trakrf-cli/abc/reply", Method: method}
+ if params != "" {
+ req.Params = json.RawMessage(params)
+ }
+ b, _ := json.Marshal(req)
+ _, out := r.d.handleRPC(b)
+ return decodeResponse(t, out)
+}
+
+func (r *powerRig) fire(t *testing.T) {
+ t.Helper()
+ require.Len(t, r.pending, 1, "exactly one delayed action should be scheduled")
+ f := r.pending[0]
+ r.pending = nil
+ f()
+}
+
+func TestReboot_RepliesThenAnnouncesThenReboots(t *testing.T) {
+ r := newPowerRig(t, true)
+
+ resp := r.call(t, readerrpc.MethodReboot, `{"requested_by":"mike@bench"}`)
+ require.Nil(t, resp.Error)
+ var res readerrpc.PowerResult
+ require.NoError(t, json.Unmarshal(resp.Result, &res))
+ assert.Equal(t, readerrpc.PowerResult{Action: readerrpc.StatusReasonReboot, InMs: 2000}, res)
+ assert.Equal(t, []time.Duration{DefaultPowerDelay}, r.delays)
+ assert.Empty(t, r.events, "nothing happens before the reply has had its delay")
+
+ r.fire(t)
+ assert.Equal(t, []string{"status offline reboot", "host reboot"}, r.events)
+}
+
+func TestShutdown_RepliesThenAnnouncesThenPowersOff(t *testing.T) {
+ r := newPowerRig(t, true)
+
+ resp := r.call(t, readerrpc.MethodShutdown, "")
+ require.Nil(t, resp.Error)
+ var res readerrpc.PowerResult
+ require.NoError(t, json.Unmarshal(resp.Result, &res))
+ assert.Equal(t, readerrpc.StatusReasonShutdown, res.Action)
+ assert.Empty(t, r.events)
+
+ r.fire(t)
+ assert.Equal(t, []string{"status offline shutdown", "host poweroff"}, r.events)
+}
+
+// The daemon's own Stop, which systemd's shutdown triggers, must keep the reason.
+func TestPower_StopKeepsTheReason(t *testing.T) {
+ r := newPowerRig(t, true)
+ r.call(t, readerrpc.MethodShutdown, "")
+ r.fire(t)
+
+ var doc readerrpc.StatusDoc
+ require.NoError(t, json.Unmarshal(r.d.offlinePayload(), &doc))
+ assert.False(t, doc.Online)
+ assert.Equal(t, readerrpc.StatusReasonShutdown, doc.Reason)
+
+ r.d.SetHealth(readerrpc.Health{API: readerrpc.APIHealthy})
+ assert.False(t, r.status[len(r.status)-1].Online, "a late health change must not announce a rebooting reader online")
+}
+
+func TestPower_SecondRequestWhilePendingIsRefused(t *testing.T) {
+ r := newPowerRig(t, true)
+ require.Nil(t, r.call(t, readerrpc.MethodReboot, "").Error)
+
+ for _, m := range []string{readerrpc.MethodReboot, readerrpc.MethodShutdown} {
+ resp := r.call(t, m, "")
+ require.NotNil(t, resp.Error, m)
+ assert.Contains(t, resp.Error.Message, "already")
+ }
+ r.fire(t)
+ assert.Equal(t, []string{"status offline reboot", "host reboot"}, r.events, "the host is asked once")
+
+ resp := r.call(t, readerrpc.MethodShutdown, "")
+ require.NotNil(t, resp.Error, "still refused once the reader is going down")
+}
+
+func TestPower_RefusedWhileAutoRecoveryRebootPending(t *testing.T) {
+ r := newPowerRig(t, true)
+ r.d.GoingDown(readerrpc.StatusReasonAutoRecovery)
+
+ resp := r.call(t, readerrpc.MethodShutdown, "")
+ require.NotNil(t, resp.Error)
+ assert.Contains(t, resp.Error.Message, readerrpc.StatusReasonAutoRecovery)
+ assert.Empty(t, r.pending)
+}
+
+// A reboot systemd refused leaves the reader up: it must not stay announced as
+// offline, and the operator must be able to try again.
+func TestPower_HostFailureGoesBackOnline(t *testing.T) {
+ r := newPowerRig(t, true)
+ r.hostErr = errors.New("systemctl reboot: exit status 1")
+ r.call(t, readerrpc.MethodReboot, "")
+ r.fire(t)
+
+ assert.Equal(t, []string{"status offline reboot", "host reboot", "status online"}, r.events)
+ last := r.status[len(r.status)-1]
+ assert.True(t, last.Online)
+ assert.Empty(t, last.Reason)
+ assert.Contains(t, r.logs.String(), "systemctl reboot: exit status 1")
+
+ r.hostErr = nil
+ require.Nil(t, r.call(t, readerrpc.MethodReboot, "").Error, "a retry is accepted")
+}
+
+func TestPower_InvalidParams(t *testing.T) {
+ r := newPowerRig(t, true)
+ resp := r.call(t, readerrpc.MethodReboot, `"now"`)
+ require.NotNil(t, resp.Error)
+ assert.Equal(t, readerrpc.CodeInvalidParams, resp.Error.Code)
+ assert.Empty(t, r.pending)
+ require.Nil(t, r.call(t, readerrpc.MethodReboot, "").Error, "a bad request does not leave the daemon thinking it is going down")
+}
+
+func TestPower_NoHostIsMethodNotFound(t *testing.T) {
+ r := newPowerRig(t, false)
+ for _, m := range []string{readerrpc.MethodReboot, readerrpc.MethodShutdown} {
+ resp := r.call(t, m, "")
+ require.NotNil(t, resp.Error, m)
+ assert.Equal(t, readerrpc.CodeMethodNotFound, resp.Error.Code)
+ }
+ assert.Empty(t, r.pending)
+
+ var caps readerrpc.Capabilities
+ require.NoError(t, json.Unmarshal(r.call(t, readerrpc.MethodGetCapabilities, "").Result, &caps))
+ assert.NotContains(t, caps.Supports, readerrpc.MethodReboot)
+ assert.NotContains(t, caps.Supports, readerrpc.MethodShutdown)
+}
+
+func TestCapabilities_WithHostListsRebootAndShutdown(t *testing.T) {
+ r := newPowerRig(t, true)
+ var caps readerrpc.Capabilities
+ require.NoError(t, json.Unmarshal(r.call(t, readerrpc.MethodGetCapabilities, "").Result, &caps))
+ assert.Contains(t, caps.Supports, readerrpc.MethodReboot)
+ assert.Contains(t, caps.Supports, readerrpc.MethodShutdown)
+ assert.NotContains(t, caps.Unsupported, readerrpc.MethodReboot)
+ assert.NotContains(t, caps.Unsupported, readerrpc.MethodShutdown)
+}
+
+func TestPower_LogsWhoAsked(t *testing.T) {
+ r := newPowerRig(t, true)
+ r.call(t, readerrpc.MethodShutdown, `{"requested_by":"mike@bench"}`)
+
+ logs := r.logs.String()
+ assert.Contains(t, logs, `"requested_by":"mike@bench"`)
+ assert.Contains(t, logs, `"src":"trakrf-cli/abc/reply"`)
+ assert.Contains(t, logs, `"method":"Reader.Shutdown"`)
+ assert.Contains(t, logs, `"asked_at"`, "the log line carries when it was asked")
+}
diff --git a/mqtt-rpc/internal/readerrpc/health.go b/mqtt-rpc/internal/readerrpc/health.go
index 48188d74..10a16275 100644
--- a/mqtt-rpc/internal/readerrpc/health.go
+++ b/mqtt-rpc/internal/readerrpc/health.go
@@ -39,6 +39,13 @@ const (
// reboots a wedged reader, so a planned recovery reads differently from a failure.
const StatusReasonAutoRecovery = "auto-recovery"
+// StatusReasonReboot and StatusReasonShutdown are the offline reasons for a
+// reboot or shutdown an operator asked for (Reader.Reboot, Reader.Shutdown).
+const (
+ StatusReasonReboot = "reboot"
+ StatusReasonShutdown = "shutdown"
+)
+
// Health is the reader-API health block of the status document.
type Health struct {
// API is one of the API* states.
@@ -76,7 +83,8 @@ type StatusDoc struct {
Online bool `json:"online"`
At string `json:"at,omitempty"`
Version string `json:"version,omitempty"`
- // Reason accompanies a planned offline (auto-recovery).
+ // Reason accompanies a planned offline: StatusReasonAutoRecovery,
+ // StatusReasonReboot or StatusReasonShutdown.
Reason string `json:"reason,omitempty"`
Health *Health `json:"health,omitempty"`
}
diff --git a/mqtt-rpc/internal/readerrpc/rpc.go b/mqtt-rpc/internal/readerrpc/rpc.go
index b1e128b9..b36920b1 100644
--- a/mqtt-rpc/internal/readerrpc/rpc.go
+++ b/mqtt-rpc/internal/readerrpc/rpc.go
@@ -22,14 +22,34 @@ const (
MethodGetOperProfile = "Reader.GetOperProfile"
MethodSetOperProfile = "Reader.SetOperProfile"
MethodGetStatus = "Reader.GetStatus"
+ MethodGpoSet = "Gpo.Set"
+
+ // MethodReboot and MethodShutdown cleanly reboot or power off the reader
+ // host (sync, then systemd). The daemon replies first and acts after a short
+ // delay, announcing the planned outage on the retained status first.
+ // Neither takes required params; PowerParams is optional.
+ MethodReboot = "Reader.Reboot"
+ MethodShutdown = "Reader.Shutdown"
// Reserved for future use.
MethodScanStart = "Scan.Start"
MethodScanStop = "Scan.Stop"
- MethodGpoSet = "Gpo.Set"
- MethodReboot = "Reader.Reboot"
)
+// PowerParams are the optional params of Reader.Reboot and Reader.Shutdown.
+type PowerParams struct {
+ // RequestedBy names the person or system asking, for the reader's log. The
+ // request's Src is logged as well.
+ RequestedBy string `json:"requested_by,omitempty"`
+}
+
+// PowerResult acknowledges Reader.Reboot or Reader.Shutdown. The action has not
+// happened yet: it starts InMs after the reply.
+type PowerResult struct {
+ Action string `json:"action"` // StatusReasonReboot or StatusReasonShutdown
+ InMs int `json:"in_ms"`
+}
+
// Standard JSON-RPC 2.0 error codes.
const (
CodeParseError = -32700