From 5ca81e0222aacce16ab763393e3eed3060ef4aa7 Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:06:00 -0600 Subject: [PATCH 1/5] feat: Reader.Reboot and Reader.Shutdown in mqtt-rpcd A clean reboot is the known fix for a wedged CS463, and customer readers have no SSH. The daemon now takes Reader.Reboot and a new Reader.Shutdown over the broker. It replies at once, waits 2 s so the reply gets out, publishes the retained status offline with reason "reboot" or "shutdown", then runs sync and systemctl reboot or poweroff. Never a power reset. A second request while one is under way, or while an automatic recovery reboot is announced, is refused. If systemd refuses, the reader is still up, so the status goes back online and a retry is accepted. Each request is logged with requested_by, the reply topic and asked_at. GetCapabilities advertises both when the daemon has a host. The CS463 adapter stops listing Reader.Reboot as unsupported: it is the daemon's. The backend's copy of rpc.go is kept identical. Refs: TRA-1316 Co-Authored-By: Claude Opus 5.5 --- backend/internal/readerrpc/rpc.go | 24 +- mqtt-rpc/cmd/mqtt-rpcd/main.go | 6 +- mqtt-rpc/internal/readerd/cs463/adapter.go | 3 +- .../internal/readerd/cs463/adapter_test.go | 3 +- mqtt-rpc/internal/readerd/daemon.go | 24 +- mqtt-rpc/internal/readerd/health_test.go | 5 + mqtt-rpc/internal/readerd/host.go | 10 + mqtt-rpc/internal/readerd/host_test.go | 11 + mqtt-rpc/internal/readerd/power.go | 101 ++++++++ mqtt-rpc/internal/readerd/power_test.go | 218 ++++++++++++++++++ mqtt-rpc/internal/readerrpc/health.go | 10 +- mqtt-rpc/internal/readerrpc/rpc.go | 24 +- 12 files changed, 427 insertions(+), 12 deletions(-) create mode 100644 mqtt-rpc/internal/readerd/power.go create mode 100644 mqtt-rpc/internal/readerd/power_test.go 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/cmd/mqtt-rpcd/main.go b/mqtt-rpc/cmd/mqtt-rpcd/main.go index fbf01209..179bfe62 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") } @@ -113,7 +117,7 @@ func main() { monitor := readerd.NewMonitor(policy, readerd.MonitorDeps{ Probe: adapter.ProbeHealth, Reconcile: reconcile, - Host: readerd.NewSystemdHost(&logger), + Host: host, Store: readerd.RebootStore{Path: readerd.DefaultRebootStorePath}, Publish: d.SetHealth, GoingDown: d.GoingDown, 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_test.go b/mqtt-rpc/internal/readerd/health_test.go index 6469c3a7..60740a21 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 } 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 From 1418c13567cd68fd7a36ddf940b6022e5cfd2c4f Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:06:55 -0600 Subject: [PATCH 2/5] fix: a refused auto-recovery reboot no longer leaves the reader announced offline The health monitor announces the reboot on the retained status, then asks systemd. When systemd refused, the daemon stayed up but its status kept saying offline: SetHealth republishes offline once a reason is set, so every later health change repeated it, for as long as the daemon ran. The monitor now calls AbortGoingDown on that failure, which clears the reason and republishes online. It retries after the next window, as before. Found while adding Reader.Reboot, which needs the same take-back. Refs: TRA-1316 Co-Authored-By: Claude Opus 5.5 --- mqtt-rpc/cmd/mqtt-rpcd/main.go | 13 ++++++------ mqtt-rpc/internal/readerd/health.go | 6 ++++++ mqtt-rpc/internal/readerd/health_test.go | 27 +++++++++++++++++++++--- 3 files changed, 37 insertions(+), 9 deletions(-) diff --git a/mqtt-rpc/cmd/mqtt-rpcd/main.go b/mqtt-rpc/cmd/mqtt-rpcd/main.go index 179bfe62..11371987 100644 --- a/mqtt-rpc/cmd/mqtt-rpcd/main.go +++ b/mqtt-rpc/cmd/mqtt-rpcd/main.go @@ -115,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: host, - 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/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 60740a21..affec8ff 100644 --- a/mqtt-rpc/internal/readerd/health_test.go +++ b/mqtt-rpc/internal/readerd/health_test.go @@ -69,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 { @@ -217,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) From 6015a59ebe7fb7c91de76a3f6717a844ba0ba2ec Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:07:39 -0600 Subject: [PATCH 3/5] feat: mqtt-rpcctl reboot and shutdown Operator commands for Reader.Reboot and Reader.Shutdown. Both take the reader offline, so both refuse without -yes, and the refusal names the cost: minutes off the broker for a reboot, and an on-site power-cycle for a shutdown. -by goes to the reader's log as requested_by and defaults to $USER@hostname. status already shows a planned outage as "offline (reboot)" or "offline (shutdown)"; tests now pin both. Refs: TRA-1316 Co-Authored-By: Claude Opus 5.5 --- mqtt-rpc/cmd/mqtt-rpcctl/main.go | 12 +++++- mqtt-rpc/cmd/mqtt-rpcctl/power.go | 54 ++++++++++++++++++++++++ mqtt-rpc/cmd/mqtt-rpcctl/power_test.go | 55 +++++++++++++++++++++++++ mqtt-rpc/cmd/mqtt-rpcctl/status_test.go | 4 ++ 4 files changed, 123 insertions(+), 2 deletions(-) create mode 100644 mqtt-rpc/cmd/mqtt-rpcctl/power.go create mode 100644 mqtt-rpc/cmd/mqtt-rpcctl/power_test.go 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"}, From 1101f0761cf78aae4bdae3678825165b0ed9f343 Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:08:16 -0600 Subject: [PATCH 4/5] docs: mqtt-rpc README covers Reader.Reboot and Reader.Shutdown The method list said Reader.Reboot was reserved and unsupported on the CS463. It now lists both methods with the daemon's own, and a new section gives the operator commands, the reply-then-act sequence, the planned-outage reasons on the status, and what happens when systemd refuses. The recovery section notes that a refused automatic reboot republishes the reader online. Refs: TRA-1316 Co-Authored-By: Claude Opus 5.5 --- mqtt-rpc/README.md | 32 ++++++++++++++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) 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} From b3d34f556ff8a85af00cf5f51d36d3e86f2b9b43 Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:08:56 -0600 Subject: [PATCH 5/5] docs: backend readerrpc copy says the status topic carries StatusDoc The backend's hand-kept copy of the contract still said Status was the status-topic payload. The daemon's copy was corrected when StatusDoc arrived; this one was not. Comment only: config.go is now identical in both copies again, as rpc.go and daemon.go are. Found while mirroring the Reader.Shutdown contract change. Refs: TRA-1316 Co-Authored-By: Claude Opus 5.5 --- backend/internal/readerrpc/config.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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"`