diff --git a/backend/internal/handlers/readerconfig/software.go b/backend/internal/handlers/readerconfig/software.go index 8622b55e..888f5272 100644 --- a/backend/internal/handlers/readerconfig/software.go +++ b/backend/internal/handlers/readerconfig/software.go @@ -15,6 +15,7 @@ import ( "github.com/trakrf/platform/backend/internal/middleware" modelerrors "github.com/trakrf/platform/backend/internal/models/errors" + "github.com/trakrf/platform/backend/internal/readercontrol" "github.com/trakrf/platform/backend/internal/readerrelease" "github.com/trakrf/platform/backend/internal/readerrpc" "github.com/trakrf/platform/backend/internal/util/httputil" @@ -156,14 +157,17 @@ func (h *Handler) UpdateSoftware(w http.ResponseWriter, r *http.Request) { res, err := h.daemon.UpdateDaemon(r.Context(), base, manifest) if err != nil { var re *readerrpc.UpdateRefusedError - if errors.As(err, &re) { - httputil.WriteJSON(w, http.StatusUnprocessableEntity, map[string]any{ - "error": "update_refused", - "reason": re.Reason, - }) - return + switch { + case errors.As(err, &re): + // The reader checked the request and changed nothing; detail is its reason. + httputil.WriteJSONError(w, r, http.StatusUnprocessableEntity, modelerrors.ErrValidation, re.Reason, reqID) + case errors.Is(err, readercontrol.ErrNoReply): + // Sent, but the reply was lost: the update may be running. 504, not + // 502, so the app watches the reader instead of saying "not started". + httputil.WriteJSONError(w, r, http.StatusGatewayTimeout, modelerrors.ErrInternal, err.Error(), reqID) + default: + httputil.WriteJSONError(w, r, http.StatusBadGateway, modelerrors.ErrInternal, err.Error(), reqID) } - httputil.WriteJSONError(w, r, http.StatusBadGateway, modelerrors.ErrInternal, err.Error(), reqID) return } httputil.WriteJSON(w, http.StatusAccepted, map[string]any{"data": res}) diff --git a/backend/internal/handlers/readerconfig/software_integration_test.go b/backend/internal/handlers/readerconfig/software_integration_test.go index fa429072..d758c011 100644 --- a/backend/internal/handlers/readerconfig/software_integration_test.go +++ b/backend/internal/handlers/readerconfig/software_integration_test.go @@ -16,6 +16,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/trakrf/platform/backend/internal/readercontrol" "github.com/trakrf/platform/backend/internal/readerrelease" "github.com/trakrf/platform/backend/internal/readerrpc" "github.com/trakrf/platform/backend/internal/testutil" @@ -90,8 +91,35 @@ func TestUpdateSoftware_RefusedIs422WithTheReason(t *testing.T) { h, org, dev := softwareFixture(t, d, &fakeReleases{m: benchManifest}) rr := do(t, h, org, http.MethodPost, fmt.Sprintf("/api/v1/scan-devices/%d/reader-software/update", dev), `{"version":"v2"}`) require.Equal(t, http.StatusUnprocessableEntity, rr.Code, rr.Body.String()) - assert.Contains(t, rr.Body.String(), "update_refused") - assert.Contains(t, rr.Body.String(), "an update is already in progress") + // The standard envelope, like every other error this API returns. + var body struct { + Error struct { + Type string `json:"type"` + Status int `json:"status"` + Detail string `json:"detail"` + } `json:"error"` + } + require.NoError(t, json.Unmarshal(rr.Body.Bytes(), &body)) + assert.Equal(t, "validation_error", body.Error.Type) + assert.Equal(t, http.StatusUnprocessableEntity, body.Error.Status) + assert.Equal(t, "an update is already in progress", body.Error.Detail) +} + +// The request reached the reader but its reply was lost: the update may well be +// running. 504 says "outcome unknown", distinct from 502 "never sent". +func TestUpdateSoftware_LostReplyIs504(t *testing.T) { + d := &fakeDaemon{updErr: fmt.Errorf("readercontrol: Daemon.Update on trakrf.id/x: %w", readercontrol.ErrNoReply)} + h, org, dev := softwareFixture(t, d, &fakeReleases{m: benchManifest}) + rr := do(t, h, org, http.MethodPost, fmt.Sprintf("/api/v1/scan-devices/%d/reader-software/update", dev), `{"version":"v2"}`) + assert.Equal(t, http.StatusGatewayTimeout, rr.Code, rr.Body.String()) + assert.Contains(t, rr.Body.String(), `"type":"internal_error"`) +} + +func TestUpdateSoftware_UnsentRequestIs502(t *testing.T) { + d := &fakeDaemon{updErr: errors.New("readercontrol: publish request: not connected")} + h, org, dev := softwareFixture(t, d, &fakeReleases{m: benchManifest}) + rr := do(t, h, org, http.MethodPost, fmt.Sprintf("/api/v1/scan-devices/%d/reader-software/update", dev), `{"version":"v2"}`) + assert.Equal(t, http.StatusBadGateway, rr.Code, rr.Body.String()) } func TestUpdateSoftware_UnknownReleaseIs404AndNeverReachesTheReader(t *testing.T) { diff --git a/backend/internal/readercontrol/client.go b/backend/internal/readercontrol/client.go index a7fb2f35..c7cac889 100644 --- a/backend/internal/readercontrol/client.go +++ b/backend/internal/readercontrol/client.go @@ -11,6 +11,7 @@ package readercontrol import ( "context" "encoding/json" + "errors" "fmt" "os" "sync" @@ -156,6 +157,11 @@ func (c *Client) next() int { return c.nextID } +// ErrNoReply: the request was published but no reply arrived in time. Unlike a +// publish failure, the reader may have received it and acted — a caller that +// started something must treat the outcome as unknown, not as "not started". +var ErrNoReply = errors.New("reader did not respond (offline?)") + // call publishes a request to the reader's /rpc topic and blocks until the // matching reply arrives, the context is canceled, or the configured timeout // elapses. On timeout the pending entry is reclaimed and an offline-style error @@ -201,12 +207,12 @@ func (c *Client) call(ctx context.Context, base, method string, params any) (rea c.mu.Lock() delete(c.pending, id) c.mu.Unlock() - return readerrpc.Response{}, fmt.Errorf("readercontrol: %s on %s: reader did not respond (offline?): %w", method, base, ctx.Err()) + return readerrpc.Response{}, fmt.Errorf("readercontrol: %s on %s: %w: %w", method, base, ErrNoReply, ctx.Err()) case <-timer.C: c.mu.Lock() delete(c.pending, id) c.mu.Unlock() - return readerrpc.Response{}, fmt.Errorf("readercontrol: %s on %s: reader did not respond (offline?)", method, base) + return readerrpc.Response{}, fmt.Errorf("readercontrol: %s on %s: %w", method, base, ErrNoReply) } } diff --git a/backend/internal/readercontrol/client_test.go b/backend/internal/readercontrol/client_test.go index 8e49d4dd..d2cac2f6 100644 --- a/backend/internal/readercontrol/client_test.go +++ b/backend/internal/readercontrol/client_test.go @@ -117,6 +117,24 @@ func TestCall_Timeout(t *testing.T) { } } +// A request that was published but never answered may still have been acted +// on: callers must be able to tell it from one that was never sent. +func TestCall_NoReplyIsErrNoReply(t *testing.T) { + c, _ := newTestClient() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, err := c.call(ctx, "trakrf.id/cs463-212", readerrpc.MethodGetStatus, nil) + if !errors.Is(err, ErrNoReply) { + t.Fatalf("context expiry: err = %v, want ErrNoReply", err) + } + + c.timeout = 20 * time.Millisecond + _, err = c.call(context.Background(), "trakrf.id/cs463-212", readerrpc.MethodGetStatus, nil) + if !errors.Is(err, ErrNoReply) { + t.Fatalf("client timeout: err = %v, want ErrNoReply", err) + } +} + func TestSetOperProfile_RPCErrorMapsToGoError(t *testing.T) { c, cp := newTestClient() diff --git a/backend/internal/readerrpc/daemon.go b/backend/internal/readerrpc/daemon.go index baa473c7..3d5fd577 100644 --- a/backend/internal/readerrpc/daemon.go +++ b/backend/internal/readerrpc/daemon.go @@ -41,6 +41,10 @@ const ( UpdateStateCommitted = "committed" UpdateStateRolledBack = "rolled_back" UpdateStateFailed = "failed" + // UpdateStateUnguarded: the target never committed, and the guard found no + // other installed version to roll back to, so the target stays live with no + // rollback armed. Unlike failed, the target DID replace the old build. + UpdateStateUnguarded = "unguarded" ) // UpdateInfo describes the most recent update attempt. diff --git a/deploy/edge/mqtt-rpc/README.md b/deploy/edge/mqtt-rpc/README.md index dabcf892..ba92a564 100644 --- a/deploy/edge/mqtt-rpc/README.md +++ b/deploy/edge/mqtt-rpc/README.md @@ -31,6 +31,10 @@ binaries keeps its old binary as version `legacy` until it ages out of retention (newest 3 versions are kept); `legacy` has no updater, so rolling back to it ends remote updates for that reader. +`install.sh` is safe to interrupt and re-run: an SSH drop at any point leaves a +runnable `/opt/trakrf/mqtt-rpcd`, and the rollback guard is stopped and disarmed +before the new binary moves in (`mqtt-rpcd-guard place`). + ## The golden CloudServer The one entity the daemon cannot create is `TrakRF mqtt-rpc MQTT Server`. It holds the broker credential and a trust bundle, so it is set up by hand when the diff --git a/deploy/edge/mqtt-rpc/install.sh b/deploy/edge/mqtt-rpc/install.sh index 1bb1899a..a1b369d7 100755 --- a/deploy/edge/mqtt-rpc/install.sh +++ b/deploy/edge/mqtt-rpc/install.sh @@ -33,16 +33,19 @@ scp "$HERE/mqtt-rpcd.env.example" "$HOST:/tmp/mqtt-rpcd.env.example" ssh "$HOST" "VERSION='$VERSION' sh -s" <<'REMOTE' set -eu cd /opt/trakrf -chmod 755 "versions/.mqtt-rpcd-$VERSION.upload" .mqtt-rpcd-guard.upload -mv -f "versions/.mqtt-rpcd-$VERSION.upload" "versions/mqtt-rpcd-$VERSION" +chmod 755 .mqtt-rpcd-guard.upload mv -f .mqtt-rpcd-guard.upload mqtt-rpcd-guard -# A pre-versioning install has a plain file here: keep it as a manual rollback -# target named "legacy" (it has no updater, so rolling back to it ends OTA). -if [ -f mqtt-rpcd ] && [ ! -L mqtt-rpcd ]; then mv -f mqtt-rpcd versions/mqtt-rpcd-legacy; fi -# An SSH install supersedes any remote update on probation. -rm -f update/pending update/starts update/ticks -rm -f .mqtt-rpcd.link -ln -s "versions/mqtt-rpcd-$VERSION" .mqtt-rpcd.link && mv -f .mqtt-rpcd.link mqtt-rpcd +# The guard's install step stops its timer. Start it again however this script +# ends — an error under set -e, or the SSH session dropping — so a failed +# install never leaves a reader with no no-commit rollback until its next boot. +restart_guard_timer() { systemctl start mqtt-rpcd-guard.timer >/dev/null 2>&1 || true; } +trap restart_guard_timer EXIT +trap 'exit 1' HUP INT TERM +# Stop the timer, disarm any remote update on probation, move the binary in, +# and swap the link — keeping a pre-versioning plain binary as "legacy" (a +# manual rollback target with no updater; rolling back to it ends OTA). Each +# step is safe to interrupt and re-run; see mqtt-rpcd-guard "place". +./mqtt-rpcd-guard place "$VERSION" test -f mqtt-rpcd.env || mv /tmp/mqtt-rpcd.env.example mqtt-rpcd.env chmod 600 mqtt-rpcd.env diff --git a/deploy/edge/mqtt-rpc/mqtt-rpcd-guard b/deploy/edge/mqtt-rpc/mqtt-rpcd-guard index af59d87d..d79ac369 100755 --- a/deploy/edge/mqtt-rpc/mqtt-rpcd-guard +++ b/deploy/edge/mqtt-rpc/mqtt-rpcd-guard @@ -18,6 +18,11 @@ # list Print the installed versions; `*` marks the active one. # rollback Operator recovery over SSH: point the link at , # drop any pending update, restart. +# place install.sh's step on the reader: stop the guard timer, +# disarm any pending update, move the uploaded binary in +# and point the link at it (a plain pre-versioning binary +# is hardlinked to "legacy" first, so the link path never +# goes missing). install.sh restarts the timer after. # recover The on-site operator's one command (the `recover` SSH # login runs it and nothing else): fetch the newest daemon # release from GitHub, check it against its manifest's @@ -113,6 +118,23 @@ resume_claim() { fi } +# fallback_version : the newest installed version (by mtime) other +# than . legacy has no updater, so it is chosen only when nothing else +# is installed. Prints nothing when there is no candidate at all. +fallback_version() { + # ls -t is the only mtime sort the reader's busybox has; every name is + # checked by valid_version before use, so odd filenames cannot leak through. + # shellcheck disable=SC2012 + c=$(ls -t "$DIR/versions" 2>/dev/null | sed -n 's/^mqtt-rpcd-//p' | while read -r cand; do + if valid_version "$cand" && [ "$cand" != "$1" ] && [ -f "$DIR/versions/mqtt-rpcd-$cand" ]; then + echo "$cand" + fi + done) + pick=$(printf '%s\n' "$c" | grep -vx legacy | head -n 1) + [ -n "$pick" ] || pick=$(printf '%s\n' "$c" | grep -x legacy | head -n 1) + echo "$pick" +} + # auto_rollback : claim the pending update and point back at `from`. # Returns 0 only when the link was moved (the caller restarts). Every failure # leaves the target in place and pending armed, so the next run retries: @@ -121,13 +143,24 @@ auto_rollback() { mv "$PENDING" "$CLAIM" 2>/dev/null || return 1 # the daemon committed first target=$(field target "$CLAIM") from=$(field from "$CLAIM") + reason="$1" if ! valid_version "$from" || [ ! -f "$DIR/versions/mqtt-rpcd-$from" ]; then - write_last failed "$target" "${1}_no_from" - log "cannot roll back $target: version '$from' is not installed ($1)" - rm -f "$CLAIM" - clear_counters - "$SYNC" - return 1 + # The recorded version is gone (removed by hand, or never recorded). + # Any other build beats leaving the failing one live with no guard. + gone="$from" + from=$(fallback_version "$target") + if [ -z "$from" ]; then + # Nothing else to run. The target stays live, unguarded; say so + # rather than "failed", which reads as "nothing replaced". + write_last unguarded "$target" "${1}_no_from" + log "cannot roll back $target: '$gone' is not installed and no other version is ($1)" + rm -f "$CLAIM" + clear_counters + "$SYNC" + return 1 + fi + reason="${1}_fallback" + log "version '$gone' is not installed; falling back to $from ($1)" fi if ! point_at "$from"; then mv "$CLAIM" "$PENDING" 2>/dev/null @@ -135,7 +168,7 @@ auto_rollback() { return 1 fi "$SYNC" # the link swap is durable before anything is deleted - write_last rolled_back "$target" "$1" + write_last rolled_back "$target" "$reason" # A build that failed probation is never a rollback target and must not push # a good version out of retention. if valid_version "$target" && [ "$target" != "$from" ]; then @@ -144,7 +177,7 @@ auto_rollback() { rm -f "$CLAIM" clear_counters "$SYNC" - log "rolled back $target -> $from ($1); removed $target" + log "rolled back $target -> $from ($reason); removed $target" return 0 } @@ -281,6 +314,46 @@ rollback) "$SYSTEMCTL" restart mqtt-rpcd exit 0 ;; +place) + # install.sh's binary-and-link step, run on the reader after it uploads + # versions/.mqtt-rpcd-.upload. Every step is safe to interrupt + # (an SSH drop) and to re-run. + v="${2:-}" + up="$DIR/versions/.mqtt-rpcd-$v.upload" + if ! valid_version "$v" || [ ! -f "$up" ]; then + log "place: no upload for version '$v' at $up" + exit 1 + fi + # No guard tick may act while the install changes what it acts on. + # install.sh starts the timer again when it finishes, however it finishes. + "$SYSTEMCTL" stop mqtt-rpcd-guard.timer 2>/dev/null || true + # An SSH install supersedes any remote update on probation. Disarm BEFORE + # the new binary moves in: a rollback of a same-version update would + # otherwise delete the binary just installed. The claim goes first, since + # resume_claim turns a lone claim back into pending. + rm -f "$CLAIM" "$PENDING.commit" "$PENDING" "$STATE/starts" "$STATE/ticks" + "$SYNC" + if ! { chmod 755 "$up" && mv -f "$up" "$DIR/versions/mqtt-rpcd-$v"; }; then + log "place: could not put $v in place" + exit 1 + fi + # A pre-versioning install has a plain binary at the link path. Keep it as + # the "legacy" rollback target by hardlink, so it never leaves the link + # path: the symlink rename below replaces it in one step. + if [ -f "$LINK" ] && [ ! -L "$LINK" ]; then + ln -f "$LINK" "$DIR/versions/mqtt-rpcd-legacy" || { + log "place: could not keep the old binary as legacy; nothing switched" + exit 1 + } + fi + if ! point_at "$v"; then + log "place: could not point the link at $v" + exit 1 + fi + "$SYNC" + log "installed $v" + exit 0 + ;; recover) mkdir -p "$STATE" : >"$RECOVER_LOG" # the last run only @@ -310,7 +383,7 @@ recover) exit "$rc" ;; *) - echo "usage: $0 start|tick|list|rollback |recover" >&2 + echo "usage: $0 start|tick|list|rollback |place |recover" >&2 exit 2 ;; esac diff --git a/frontend/src/components/scandevices/ReaderSoftwarePanel.test.tsx b/frontend/src/components/scandevices/ReaderSoftwarePanel.test.tsx index 953c14ba..b1bce694 100644 --- a/frontend/src/components/scandevices/ReaderSoftwarePanel.test.tsx +++ b/frontend/src/components/scandevices/ReaderSoftwarePanel.test.tsx @@ -126,9 +126,13 @@ describe('ReaderSoftwarePanel', () => { startError: { response: { status: 422, + // The standard error envelope. data: { - error: 'update_refused', - reason: 'an update is already in progress', + error: { + type: 'validation_error', + status: 422, + detail: 'an update is already in progress', + }, }, }, }, @@ -136,6 +140,32 @@ describe('ReaderSoftwarePanel', () => { expect(screen.getByText(/an update is already in progress/)).toBeInTheDocument(); }); + it('does not say "nothing replaced" for a target left running unguarded', () => { + setup({ + software: { + daemon: { + version: 'v2', + update: { state: 'unguarded', target: 'v2', error: 'no_commit_no_from' }, + }, + supports_update: true, + }, + }); + expect(screen.getByText(/v2 is running but never confirmed/i)).toBeInTheDocument(); + expect(screen.queryByText(/nothing replaced/i)).not.toBeInTheDocument(); + }); + + it('explains an unguarded outcome while watching it land', () => { + setup({ + software: { + daemon: { version: 'v2', update: { state: 'unguarded', target: 'v2' } }, + supports_update: true, + }, + target: 'v2', + outcome: 'unguarded', + }); + expect(screen.getByText(/no earlier version to roll back to/i)).toBeInTheDocument(); + }); + it('tells a non-admin why the update was not started', () => { setup({ software: { daemon: { version: 'v1' }, supports_update: true }, diff --git a/frontend/src/components/scandevices/ReaderSoftwarePanel.tsx b/frontend/src/components/scandevices/ReaderSoftwarePanel.tsx index 734c4a4f..b98fad95 100644 --- a/frontend/src/components/scandevices/ReaderSoftwarePanel.tsx +++ b/frontend/src/components/scandevices/ReaderSoftwarePanel.tsx @@ -21,6 +21,8 @@ function lastUpdateText(u: DaemonUpdateInfo): string { return `Last update: rolled back from ${u.target ?? 'an update'}${why}`; case 'failed': return `Last update: ${u.target ?? 'update'} failed, nothing replaced${why}`; + case 'unguarded': + return `Last update: ${u.target ?? 'the update'} is running but never confirmed it came online, and no earlier version was left to roll back to${why}`; case 'pending': return `Update ${u.target ?? ''} is running and has not committed yet`; default: @@ -32,11 +34,11 @@ function lastUpdateText(u: DaemonUpdateInfo): string { function startErrorText(err: unknown): string | null { if (!err) return null; const e = err as { - response?: { status?: number; data?: { reason?: string; detail?: string } }; + response?: { status?: number; data?: { error?: { detail?: string } } }; }; const status = e.response?.status; - if (status === 422 && e.response?.data?.reason) - return `The reader refused: ${e.response.data.reason}`; + const detail = e.response?.data?.error?.detail; + if (status === 422 && detail) return `The reader refused: ${detail}`; if (status === 403) return 'Only an org admin can update reader software.'; if (status === 404) return 'No release with that version.'; return 'Could not start the update. Is the reader online?'; @@ -147,6 +149,12 @@ export function ReaderSoftwarePanel({ deviceId }: ReaderSoftwarePanelProps) { {daemon?.update?.error ? `: ${daemon.update.error}` : '.'}

)} + {outcome === 'unguarded' && ( +

+ {target} never confirmed it came online, and the reader had no earlier version to roll + back to, so it is still running {target}. Check the reader, then update it again. +

+ )} {outcome === 'timeout' && (

No answer from the reader after the update to {target}. Check that it is powered and diff --git a/frontend/src/hooks/scandevices/useReaderSoftware.test.ts b/frontend/src/hooks/scandevices/useReaderSoftware.test.ts index 6ba6dea4..5aaab368 100644 --- a/frontend/src/hooks/scandevices/useReaderSoftware.test.ts +++ b/frontend/src/hooks/scandevices/useReaderSoftware.test.ts @@ -12,6 +12,7 @@ import { import { scanDevicesApi } from '@/lib/api/scandevices'; import type { DaemonUpdateInfo, ReaderSoftware } from '@/types/scandevices'; +vi.mock('@/lib/auth/orgContext', () => ({ ensureOrgContext: vi.fn(() => Promise.resolve()) })); vi.mock('@/lib/api/scandevices', () => ({ scanDevicesApi: { getReaderSoftware: vi.fn(), startReaderUpdate: vi.fn() }, })); @@ -75,6 +76,26 @@ describe('updateOutcome', () => { expect(updateOutcome('v2', sw('v1'), t0, t0 + UPDATE_WATCH_MS + 1)).toBe('timeout'); expect(UPDATE_WATCH_MS).toBeGreaterThanOrEqual(20 * 60 * 1000); }); + + it('does not time out on a clock that ran while nobody checked the reader', () => { + // A background tab: the clock kept running, refetches paused. The last + // check was long before the deadline, so the reader has not been asked. + const late = t0 + UPDATE_WATCH_MS + 60_000; + expect(updateOutcome('v2', sw('v1'), t0, late, t0 + 1000, t0 + 1000)).toBe('watching'); + // A check after the deadline — even a failed one — settles it. + expect(updateOutcome('v2', sw('v1'), t0, late, t0 + 1000, late)).toBe('timeout'); + }); + + it('reports a target left running with nothing to roll back to', () => { + expect( + updateOutcome( + 'v2', + sw('v2', { state: 'unguarded', target: 'v2', error: 'no_commit_no_from' }), + t0, + t0 + 1000 + ) + ).toBe('unguarded'); + }); }); describe('readerUpdateInFlight', () => { @@ -121,4 +142,38 @@ describe('useReaderSoftware', () => { await vi.advanceTimersByTimeAsync(UPDATE_POLL_MS * 3); expect(get.mock.calls.length).toBe(calls); }); + + function renderSoftware() { + vi.mocked(scanDevicesApi.getReaderSoftware).mockReturnValue( + Promise.resolve({ data: { data: sw('v1') } } as never) + ); + const qc = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + const wrapper = ({ children }: { children: ReactNode }) => + createElement(QueryClientProvider, { client: qc }, children); + return renderHook(() => useReaderSoftware(10), { wrapper }); + } + + it('watches the reader when the start request reached it but the reply was lost', async () => { + // 504: Daemon.Update was sent and may be running. Saying "could not start" + // would be wrong, and not watching would miss its outcome. + vi.mocked(scanDevicesApi.startReaderUpdate).mockRejectedValue({ response: { status: 504 } }); + const { result } = renderSoftware(); + await waitFor(() => expect(result.current.software).toBeDefined()); + + await result.current.startUpdate('v2').catch(() => undefined); + await waitFor(() => expect(result.current.target).toBe('v2')); + expect(result.current.outcome).toBe('watching'); + expect(result.current.startError).toBeNull(); + }); + + it('does not watch an update that was never sent', async () => { + vi.mocked(scanDevicesApi.startReaderUpdate).mockRejectedValue({ response: { status: 502 } }); + const { result } = renderSoftware(); + await waitFor(() => expect(result.current.software).toBeDefined()); + + await result.current.startUpdate('v2').catch(() => undefined); + await waitFor(() => expect(result.current.startError).not.toBeNull()); + expect(result.current.target).toBeNull(); + expect(result.current.outcome).toBeNull(); + }); }); diff --git a/frontend/src/hooks/scandevices/useReaderSoftware.ts b/frontend/src/hooks/scandevices/useReaderSoftware.ts index b70bb661..30ac6a29 100644 --- a/frontend/src/hooks/scandevices/useReaderSoftware.ts +++ b/frontend/src/hooks/scandevices/useReaderSoftware.ts @@ -14,7 +14,13 @@ export const UPDATE_POLL_MS = 5_000; */ export const UPDATE_WATCH_MS = 25 * 60 * 1000; -export type UpdateOutcome = 'watching' | 'done' | 'rolled_back' | 'failed' | 'timeout'; +export type UpdateOutcome = + | 'watching' + | 'done' + | 'rolled_back' + | 'failed' + | 'unguarded' + | 'timeout'; /** * Where an update to `target`, started at `startedAt`, stands given the latest @@ -23,13 +29,18 @@ export type UpdateOutcome = 'watching' | 'done' | 'rolled_back' | 'failed' | 'ti * rollback is not mistaken for this one's — and only a report fetched after the * update started (`fetchedAt`), so a retry of the same version is not judged by * the previous attempt's cached result. + * + * `checkedAt` is the last time the reader was asked, answered or not. Timeout + * needs a check after the deadline: in a background tab the clock runs on while + * refetches pause, and a deadline that passed unobserved proves nothing. */ export function updateOutcome( target: string, software: ReaderSoftware | undefined, startedAt: number, now: number, - fetchedAt: number = Number.POSITIVE_INFINITY + fetchedAt: number = Number.POSITIVE_INFINITY, + checkedAt: number = now ): UpdateOutcome { const fresh = fetchedAt > startedAt; const daemon = fresh ? software?.daemon : undefined; @@ -37,7 +48,9 @@ export function updateOutcome( if (daemon?.version === target && update?.state === 'committed') return 'done'; if (update?.target === target && update.state === 'rolled_back') return 'rolled_back'; if (update?.target === target && update.state === 'failed') return 'failed'; - if (now - startedAt > UPDATE_WATCH_MS) return 'timeout'; + if (update?.target === target && update.state === 'unguarded') return 'unguarded'; + if (now - startedAt > UPDATE_WATCH_MS && checkedAt - startedAt > UPDATE_WATCH_MS) + return 'timeout'; return 'watching'; } @@ -51,6 +64,11 @@ export function readerUpdateInFlight(software: ReaderSoftware | undefined): bool return state === 'downloading' || state === 'installing' || state === 'pending'; } +/** The start request was sent but its reply was lost (backend 504). */ +function replyLost(err: unknown): boolean { + return (err as { response?: { status?: number } } | null)?.response?.status === 504; +} + interface Watch { target: string; startedAt: number; @@ -80,7 +98,14 @@ export function useReaderSoftware(deviceId: number | null) { const outcome: UpdateOutcome | null = watch ? (watch.settled ?? - updateOutcome(watch.target, query.data, watch.startedAt, now, query.dataUpdatedAt)) + updateOutcome( + watch.target, + query.data, + watch.startedAt, + now, + query.dataUpdatedAt, + Math.max(query.dataUpdatedAt, query.errorUpdatedAt) + )) : null; useEffect(() => { @@ -105,6 +130,16 @@ export function useReaderSoftware(deviceId: number | null) { setNow(startedAt); setWatch({ target: res.to, startedAt }); }, + onError: (err, version) => { + // 504: the request reached the reader but its reply was lost, so the + // update may be running. Watch for its outcome rather than report it as + // not started; a reader that never got it settles as a timeout. + if (replyLost(err)) { + const startedAt = Date.now(); + setNow(startedAt); + setWatch({ target: version, startedAt }); + } + }, }); return { @@ -115,7 +150,7 @@ export function useReaderSoftware(deviceId: number | null) { await mutation.mutateAsync(version); }, isStarting: mutation.isPending, - startError: mutation.error as unknown, + startError: replyLost(mutation.error) ? null : (mutation.error as unknown), target: watch?.target ?? null, outcome, /** The reader reports an update in flight, whoever started it. */ diff --git a/frontend/src/types/scandevices/index.ts b/frontend/src/types/scandevices/index.ts index c1692520..dd759cab 100644 --- a/frontend/src/types/scandevices/index.ts +++ b/frontend/src/types/scandevices/index.ts @@ -154,7 +154,10 @@ export type DaemonUpdateState = | 'pending' | 'committed' | 'rolled_back' - | 'failed'; + | 'failed' + // The target never committed and the reader had no other version to roll + // back to, so the target is still running with no rollback armed. + | 'unguarded'; export interface DaemonUpdateInfo { state: DaemonUpdateState; diff --git a/mqtt-rpc/README.md b/mqtt-rpc/README.md index 83263af8..d946548b 100644 --- a/mqtt-rpc/README.md +++ b/mqtt-rpc/README.md @@ -336,10 +336,19 @@ reconcile: `docs/adr/0024-a-remote-update-is-undone-by-what-it-cannot-replace.md `pending` stays armed for the next run; a rollback interrupted mid-way resumes. Commit and rollback both claim `pending` by rename, so exactly one wins. + - If `from` is no longer installed, it falls back to the newest other + installed version (`legacy` only as a last resort). With nothing else + installed the failed build stays live, unguarded. `update/last` is reported in `GetCapabilities().daemon.update`: -`committed`, `rolled_back` (`error`: `crash_loop` | `no_commit` | `manual`) or -`failed` (nothing replaced). The updater never reads or writes +`committed`, `rolled_back` (`error`: `crash_loop` | `no_commit` | `manual`, +with `_fallback` when it fell back past a missing `from`), `unguarded` +(`_no_from`: the target is live and nothing was left to roll back to) or +`failed` (nothing replaced). + +Downloads trust the reader's CA bundle plus Mozilla roots embedded in the +binary (`golang.org/x/crypto/x509roots/fallback`), so every update refreshes +them; the firmware bundle itself is never updated. The updater never reads or writes `/opt/trakrf/mqtt-rpcd.env` — it holds the only on-device copy of the reader password. diff --git a/mqtt-rpc/go.mod b/mqtt-rpc/go.mod index 935833ed..50ed744d 100644 --- a/mqtt-rpc/go.mod +++ b/mqtt-rpc/go.mod @@ -1,11 +1,12 @@ module github.com/trakrf/platform/mqtt-rpc -go 1.25.1 +go 1.26.0 require ( github.com/eclipse/paho.mqtt.golang v1.5.0 github.com/rs/zerolog v1.33.0 github.com/stretchr/testify v1.11.1 + golang.org/x/crypto/x509roots/fallback v0.0.0-20261105095718-a1058eb15227 ) require ( diff --git a/mqtt-rpc/go.sum b/mqtt-rpc/go.sum index e381cd5b..ca53046a 100644 --- a/mqtt-rpc/go.sum +++ b/mqtt-rpc/go.sum @@ -19,6 +19,8 @@ github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8= github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +golang.org/x/crypto/x509roots/fallback v0.0.0-20261105095718-a1058eb15227 h1:/krupS5JpL2HJMg0bGYrXh1Zm4HXOexqCsevjyUbomk= +golang.org/x/crypto/x509roots/fallback v0.0.0-20261105095718-a1058eb15227/go.mod h1:HPze8vhfG6fO06AM+VSvxRm4E3+5Yk375mgrJ5M2z1E= golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys= golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE= golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= diff --git a/mqtt-rpc/internal/readerd/cs463/servlet_test.go b/mqtt-rpc/internal/readerd/cs463/servlet_test.go index 7290dca2..efed9f7e 100644 --- a/mqtt-rpc/internal/readerd/cs463/servlet_test.go +++ b/mqtt-rpc/internal/readerd/cs463/servlet_test.go @@ -315,7 +315,7 @@ func TestSetProfilePowerExtraBankWordCountAtLeastOne(t *testing.T) { defer f.close() fields := sampleProfileFields() - fields["firstExtraBankLength"] = "0" // as a real reader profile carries it + fields["firstExtraBankLength"] = "0" // as a real reader profile carries it fields["secondExtraBankLength"] = "0" err := f.client().SetProfilePower(context.Background(), NameProfile, 4, diff --git a/mqtt-rpc/internal/readerd/update.go b/mqtt-rpc/internal/readerd/update.go index dd5f3f96..c10a45bb 100644 --- a/mqtt-rpc/internal/readerd/update.go +++ b/mqtt-rpc/internal/readerd/update.go @@ -6,6 +6,7 @@ import ( "time" "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc" + "github.com/trakrf/platform/mqtt-rpc/internal/update" ) // selfTestInterval is how often a daemon on probation re-sends its loopback RPC. @@ -81,6 +82,9 @@ func (d *Daemon) answered() { return // the self-test keeps answering every 30 s; a later one commits } if err := d.updater.Commit(); err != nil { + if errors.Is(err, update.ErrAlreadyCommitted) { + return // a reply racing this one already committed + } d.log.Error().Err(err).Msg("update commit failed; the guard will roll back if this persists") return } diff --git a/mqtt-rpc/internal/readerd/update_test.go b/mqtt-rpc/internal/readerd/update_test.go index a2fb181e..fd0d3d1a 100644 --- a/mqtt-rpc/internal/readerd/update_test.go +++ b/mqtt-rpc/internal/readerd/update_test.go @@ -1,21 +1,25 @@ package readerd import ( + "bytes" "encoding/json" "sync" "testing" "time" + "github.com/rs/zerolog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc" + "github.com/trakrf/platform/mqtt-rpc/internal/update" ) type fakeUpdater struct { mu sync.Mutex probation bool commits int + commitErr error started []readerrpc.DaemonUpdateParams startErr error info readerrpc.DaemonInfo @@ -41,7 +45,7 @@ func (f *fakeUpdater) Commit() error { defer f.mu.Unlock() f.commits++ f.probation = false - return nil + return f.commitErr } func (f *fakeUpdater) commitCount() int { f.mu.Lock() @@ -188,6 +192,23 @@ func TestProbation_NoCommitBeforeTheMinimumUptime(t *testing.T) { assert.Less(t, DefaultCommitAfter, 15*time.Minute, "well inside the guard's 15-20 min no-commit window") } +// Two replies race through answered(): both saw probation, and the loser's +// Commit reports the winner's commit. That is not a failure and must not be +// logged as one. +func TestAnswered_LosingACommitRaceToItselfIsNotAnError(t *testing.T) { + var buf bytes.Buffer + log := zerolog.New(&buf) + d := New(Config{Broker: BrokerConfig{BaseTopic: "trakrf.id/cs463-212"}}, &fakeAdapter{}, &log) + u := &fakeUpdater{probation: true, commitErr: update.ErrAlreadyCommitted} + d.SetUpdater(u) + d.now = func() time.Time { return d.started.Add(DefaultCommitAfter) } + + d.answered() + assert.Equal(t, 1, u.commitCount()) + assert.NotContains(t, buf.String(), "commit failed") + assert.NotContains(t, buf.String(), `"level":"error"`) +} + func TestNoCommitWhenReplyPublishFails(t *testing.T) { d := newTestDaemon(&fakeAdapter{}) u := &fakeUpdater{probation: true} diff --git a/mqtt-rpc/internal/readerrpc/daemon.go b/mqtt-rpc/internal/readerrpc/daemon.go index baa473c7..3d5fd577 100644 --- a/mqtt-rpc/internal/readerrpc/daemon.go +++ b/mqtt-rpc/internal/readerrpc/daemon.go @@ -41,6 +41,10 @@ const ( UpdateStateCommitted = "committed" UpdateStateRolledBack = "rolled_back" UpdateStateFailed = "failed" + // UpdateStateUnguarded: the target never committed, and the guard found no + // other installed version to roll back to, so the target stays live with no + // rollback armed. Unlike failed, the target DID replace the old build. + UpdateStateUnguarded = "unguarded" ) // UpdateInfo describes the most recent update attempt. diff --git a/mqtt-rpc/internal/update/guard_fallback_test.go b/mqtt-rpc/internal/update/guard_fallback_test.go new file mode 100644 index 00000000..c8aab1ca --- /dev/null +++ b/mqtt-rpc/internal/update/guard_fallback_test.go @@ -0,0 +1,80 @@ +package update + +// When the version an update would roll back to is gone, the guard falls back to +// any other installed version rather than leave the failing build live with no +// guard. Only when there is nothing at all does it record "unguarded". + +import ( + "os" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc" +) + +// installedAged installs versions with mtimes one hour apart, newest first. +func installedAged(t *testing.T, active string, others ...string) Layout { + t.Helper() + l := installed(t, active, others...) + for i, v := range append([]string{active}, others...) { + at := time.Now().Add(-time.Duration(i) * time.Hour) + require.NoError(t, os.Chtimes(l.VersionPath(v), at, at)) + } + return l +} + +func armedForRollback(t *testing.T, l Layout, target, from string) { + t.Helper() + require.NoError(t, WritePending(l, Pending{Target: target, From: from})) + require.NoError(t, os.WriteFile(l.StateDir()+"/ticks", []byte("3\n"), 0o644)) +} + +func TestGuard_MissingFromFallsBackToTheNewestOtherVersion(t *testing.T) { + l := installedAged(t, "v3", "v2", "v1") + armedForRollback(t, l, "v3", "v0") + + _, code, calls := runGuard(t, l, "tick") + require.Equal(t, 0, code) + assert.Equal(t, "v2", activeOf(t, l)) + assert.Equal(t, "restart mqtt-rpcd\n", calls) + assert.NoFileExists(t, l.VersionPath("v3"), "the build that failed probation is removed") + assert.FileExists(t, l.VersionPath("v1")) + _, ok, _ := ReadPending(l) + assert.False(t, ok) + last, _ := ReadLast(l) + assert.Equal(t, readerrpc.UpdateStateRolledBack, last.State) + assert.Equal(t, "v3", last.Target) + assert.Equal(t, "no_commit_fallback", last.Error) +} + +// legacy has no updater, so it is the last resort — but a daemon that reads tags +// beats a build that never committed. +func TestGuard_MissingFromPrefersAnyVersionOverLegacy(t *testing.T) { + l := installedAged(t, "v3", "legacy", "v1") + armedForRollback(t, l, "v3", "v0") + _, code, _ := runGuard(t, l, "tick") + require.Equal(t, 0, code) + assert.Equal(t, "v1", activeOf(t, l)) + + m := installedAged(t, "v3", "legacy") + armedForRollback(t, m, "v3", "v0") + _, code, _ = runGuard(t, m, "tick") + require.Equal(t, 0, code) + assert.Equal(t, "legacy", activeOf(t, m)) +} + +func TestGuard_CrashLoopWithMissingFromFallsBackToo(t *testing.T) { + l := installedAged(t, "v3", "v2") + require.NoError(t, WritePending(l, Pending{Target: "v3", From: "v0"})) + require.NoError(t, os.WriteFile(l.StateDir()+"/starts", []byte("5\n"), 0o644)) + + _, code, calls := runGuard(t, l, "start") + require.Equal(t, 0, code) + assert.Equal(t, "v2", activeOf(t, l)) + assert.Empty(t, calls, "start runs before systemd execs the link: no restart") + last, _ := ReadLast(l) + assert.Equal(t, "crash_loop_fallback", last.Error) +} diff --git a/mqtt-rpc/internal/update/guard_place_test.go b/mqtt-rpc/internal/update/guard_place_test.go new file mode 100644 index 00000000..8f016476 --- /dev/null +++ b/mqtt-rpc/internal/update/guard_place_test.go @@ -0,0 +1,124 @@ +package update + +// `mqtt-rpcd-guard place ` is install.sh's binary-and-link step, run +// on the reader over SSH. It lives in the guard so these tests can run it: an +// SSH drop at any instant must leave a runnable /opt/trakrf/mqtt-rpcd, and no +// guard may act on the install while it happens. + +import ( + "os" + "path/filepath" + "syscall" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// upload stages versions/.mqtt-rpcd-.upload as install.sh's scp does. +func upload(t *testing.T, l Layout, v string) { + t.Helper() + require.NoError(t, os.MkdirAll(l.VersionsDir(), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(l.VersionsDir(), ".mqtt-rpcd-"+v+".upload"), []byte("binary "+v), 0o644)) +} + +// snapshotStub is a systemctl stub that records, with each call, whether +// pending and the new binary existed at that moment. +func snapshotStub(t *testing.T, l Layout, v string) []string { + t.Helper() + bin := l.Dir + "-snap" + require.NoError(t, os.MkdirAll(bin, 0o755)) + stub := filepath.Join(bin, "systemctl") + script := "#!/bin/sh\n" + + "p=no; [ -f '" + l.Pending() + "' ] && p=yes\n" + + "b=no; [ -f '" + l.VersionPath(v) + "' ] && b=yes\n" + + "echo \"$* pending=$p binary=$b\" >> '" + filepath.Join(bin, "calls") + "'\n" + require.NoError(t, os.WriteFile(stub, []byte(script), 0o755)) + return []string{"GUARD_SYSTEMCTL=" + stub} +} + +func snapshotCalls(t *testing.T, l Layout) string { + t.Helper() + b, _ := os.ReadFile(filepath.Join(l.Dir+"-snap", "calls")) + return string(b) +} + +func inode(t *testing.T, path string) uint64 { + t.Helper() + fi, err := os.Stat(path) + require.NoError(t, err) + return fi.Sys().(*syscall.Stat_t).Ino +} + +func TestGuardPlace_MigratesAPlainBinaryWithNoWindow(t *testing.T) { + l := Layout{Dir: t.TempDir()} + require.NoError(t, os.WriteFile(l.Link(), []byte("plain"), 0o755)) + plain := inode(t, l.Link()) + upload(t, l, "v2") + + out, code, _ := runGuard(t, l, "place", "v2") + require.Equal(t, 0, code, out) + + assert.Equal(t, "v2", activeOf(t, l)) + // A hardlink, not a move: the plain file stayed at the link path until the + // symlink was renamed over it. + assert.Equal(t, plain, inode(t, l.VersionPath("legacy"))) + got, _ := os.ReadFile(l.Link()) + assert.Equal(t, "binary v2", string(got)) + fi, _ := os.Stat(l.VersionPath("v2")) + assert.NotZero(t, fi.Mode()&0o111) + assert.NoFileExists(t, filepath.Join(l.VersionsDir(), ".mqtt-rpcd-v2.upload")) +} + +// An SSH drop after the hardlink, before the swap: the plain binary is still in +// place, and running install again converges. +func TestGuardPlace_ReRunAfterAnInterruptedMigrationConverges(t *testing.T) { + l := Layout{Dir: t.TempDir()} + require.NoError(t, os.WriteFile(l.Link(), []byte("plain"), 0o755)) + require.NoError(t, os.MkdirAll(l.VersionsDir(), 0o755)) + require.NoError(t, os.Link(l.Link(), l.VersionPath("legacy"))) + upload(t, l, "v2") + + out, code, _ := runGuard(t, l, "place", "v2") + require.Equal(t, 0, code, out) + assert.Equal(t, "v2", activeOf(t, l)) + legacy, _ := os.ReadFile(l.VersionPath("legacy")) + assert.Equal(t, "plain", string(legacy)) +} + +func TestGuardPlace_StopsTheTimerAndDisarmsBeforeTheBinaryMovesIn(t *testing.T) { + l := installed(t, "v2", "v1") + require.NoError(t, WritePending(l, Pending{Target: "v2", From: "v1"})) + for _, f := range []string{"starts", "ticks", "rollback.claim", "pending.commit"} { + require.NoError(t, os.WriteFile(filepath.Join(l.StateDir(), f), []byte("1\n"), 0o644)) + } + // Reinstalling the very version on probation: a guard tick after the binary + // moved in would roll back and delete the just-installed v2. + require.NoError(t, os.Remove(l.VersionPath("v2"))) + upload(t, l, "v2") + + out, code, _ := runGuardEnv(t, l, snapshotStub(t, l, "v2"), "place", "v2") + require.Equal(t, 0, code, out) + + assert.Equal(t, "stop mqtt-rpcd-guard.timer pending=yes binary=no\n", snapshotCalls(t, l), + "the timer stops first, while nothing has changed yet") + for _, f := range []string{"pending", "starts", "ticks", "rollback.claim", "pending.commit"} { + assert.NoFileExists(t, filepath.Join(l.StateDir(), f)) + } + assert.Equal(t, "v2", activeOf(t, l)) + got, _ := os.ReadFile(l.Link()) + assert.Equal(t, "binary v2", string(got)) + assert.FileExists(t, l.VersionPath("v1")) +} + +func TestGuardPlace_RefusesWithoutAnUploadOrAValidVersion(t *testing.T) { + l := installed(t, "v1") + _, code, calls := runGuard(t, l, "place", "v2") + assert.NotEqual(t, 0, code) + assert.Empty(t, calls, "nothing stopped for an install that cannot happen") + assert.Equal(t, "v1", activeOf(t, l)) + + _, code, _ = runGuard(t, l, "place", "../x") + assert.NotEqual(t, 0, code) + assert.Equal(t, "v1", activeOf(t, l)) +} diff --git a/mqtt-rpc/internal/update/guard_test.go b/mqtt-rpc/internal/update/guard_test.go index 8f2917ef..c2f5ccd0 100644 --- a/mqtt-rpc/internal/update/guard_test.go +++ b/mqtt-rpc/internal/update/guard_test.go @@ -127,7 +127,9 @@ func TestGuard_MissingFromClearsWithoutSwap(t *testing.T) { _, ok, _ := ReadPending(l) assert.False(t, ok) last, _ := ReadLast(l) - assert.Equal(t, readerrpc.UpdateStateFailed, last.State) + // Not "failed": that reads as "nothing replaced", and v2 is live. + assert.Equal(t, readerrpc.UpdateStateUnguarded, last.State) + assert.Equal(t, "v2", last.Target) assert.Equal(t, "no_commit_no_from", last.Error) } diff --git a/mqtt-rpc/internal/update/hardening_test.go b/mqtt-rpc/internal/update/hardening_test.go new file mode 100644 index 00000000..c9e831ee --- /dev/null +++ b/mqtt-rpc/internal/update/hardening_test.go @@ -0,0 +1,78 @@ +package update + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/trakrf/platform/mqtt-rpc/internal/readerrpc" +) + +// A timer unit that exists but is not enabled never ticks, so an update would +// run with no no-commit rollback. systemd records "enabled" as the wants link. +func TestPreflight_RefusesATimerThatIsNotEnabled(t *testing.T) { + f := newFixture(t) + require.NoError(t, os.Remove(guardTimerWants(f.u.GuardTimerUnit))) + _, err := f.u.Start(f.params("v2", "/mqtt-rpcd")) + refused(t, err) + assert.Contains(t, err.Error(), "not enabled") + assertStillOnV1(t, f.l) +} + +func TestGuardTimerWants_IsTheTimersTargetWantsLink(t *testing.T) { + assert.Equal(t, "/etc/systemd/system/timers.target.wants/mqtt-rpcd-guard.timer", + guardTimerWants(DefaultGuardTimerUnit)) +} + +func ageFile(t *testing.T, path string, age time.Duration) { + t.Helper() + require.NoError(t, os.WriteFile(path, []byte("partial"), 0o644)) + old := time.Now().Add(-age) + require.NoError(t, os.Chtimes(path, old, old)) +} + +// A kill or power cut mid-download leaves the hidden temp file behind, and Prune +// skips hidden files, so nothing else would ever remove it. +func TestCleanStale_RemovesOnlyAbandonedTempFiles(t *testing.T) { + l := installed(t, "v1") + dir := l.VersionsDir() + oldDownload := filepath.Join(dir, ".mqtt-rpcd-v2.download") + oldUpload := filepath.Join(dir, ".mqtt-rpcd-v4.upload") + live := filepath.Join(dir, ".mqtt-rpcd-v3.download") + other := filepath.Join(dir, ".something-else") + ageFile(t, oldDownload, time.Hour) + ageFile(t, oldUpload, time.Hour) + ageFile(t, live, time.Minute) + ageFile(t, other, time.Hour) + + require.NoError(t, CleanStale(l, staleDownloadAge)) + + assert.NoFileExists(t, oldDownload) + assert.NoFileExists(t, oldUpload) + assert.FileExists(t, live, "a download still in progress is never removed") + assert.FileExists(t, other, "only the updater's own temp files are touched") + assert.FileExists(t, l.VersionPath("v1")) +} + +func TestCheckPending_CleansAbandonedDownloads(t *testing.T) { + f := newFixture(t) + stale := filepath.Join(f.l.VersionsDir(), ".mqtt-rpcd-v9.download") + ageFile(t, stale, time.Hour) + require.NoError(t, f.u.CheckPending()) + assert.NoFileExists(t, stale) +} + +// Two replies racing through answered() both see probation, then both call +// Commit: the loser must learn "already committed", not a failure. +func TestCommit_SecondCommitIsAlreadyCommitted(t *testing.T) { + f := newFixture(t) + require.NoError(t, WritePending(f.l, Pending{Target: "v1", From: "v0"})) + require.NoError(t, f.u.CheckPending()) + require.NoError(t, f.u.Commit()) + assert.ErrorIs(t, f.u.Commit(), ErrAlreadyCommitted) + assert.Equal(t, readerrpc.UpdateStateCommitted, f.u.Info().Update.State) +} diff --git a/mqtt-rpc/internal/update/layout.go b/mqtt-rpc/internal/update/layout.go index a3e770af..6bb64447 100644 --- a/mqtt-rpc/internal/update/layout.go +++ b/mqtt-rpc/internal/update/layout.go @@ -7,6 +7,7 @@ import ( "path/filepath" "sort" "strings" + "time" ) // DefaultDir is the reader install directory. @@ -19,6 +20,12 @@ const DefaultGuardTimerUnit = "/etc/systemd/system/mqtt-rpcd-guard.timer" const binName = "mqtt-rpcd" +// guardTimerWants is the link `systemctl enable` creates for the guard timer +// unit: its presence is what "enabled" means. +func guardTimerWants(unit string) string { + return filepath.Join(filepath.Dir(unit), "timers.target.wants", filepath.Base(unit)) +} + // Layout names every path of an install. On a reader: // // /opt/trakrf/mqtt-rpcd -> versions/mqtt-rpcd- (symlink, systemd ExecStart) @@ -79,7 +86,7 @@ func ActiveVersion(l Layout) (string, error) { } // Prune keeps the newest keep versions (by mtime) plus every protected version, -// and removes the rest. Hidden in-progress downloads are left alone. +// and removes the rest. Hidden temp files are CleanStale's. func Prune(l Layout, keep int, protect ...string) error { entries, err := os.ReadDir(l.VersionsDir()) if err != nil { @@ -117,6 +124,35 @@ func Prune(l Layout, keep int, protect ...string) error { return errors.Join(errs...) } +// CleanStale removes the updater's own hidden temp files in versions/ — +// .mqtt-rpcd-.download (this daemon and the guard's recover) and +// .mqtt-rpcd-.upload (install.sh) — once they are older than olderThan. An +// interrupted transfer leaves one behind and nothing else ever removes it; the +// age keeps a transfer still in progress safe. +func CleanStale(l Layout, olderThan time.Duration) error { + entries, err := os.ReadDir(l.VersionsDir()) + if err != nil { + return err + } + cutoff := time.Now().Add(-olderThan) + var errs []error + for _, e := range entries { + n := e.Name() + if e.IsDir() || !strings.HasPrefix(n, "."+binName+"-") || + (!strings.HasSuffix(n, ".download") && !strings.HasSuffix(n, ".upload")) { + continue + } + info, err := e.Info() + if err != nil || info.ModTime().After(cutoff) { + continue + } + if err := os.Remove(filepath.Join(l.VersionsDir(), n)); err != nil && !errors.Is(err, os.ErrNotExist) { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + // syncDir fsyncs a directory so a rename in it survives a power cut. func syncDir(dir string) error { d, err := os.Open(dir) diff --git a/mqtt-rpc/internal/update/roots.go b/mqtt-rpc/internal/update/roots.go new file mode 100644 index 00000000..8e166195 --- /dev/null +++ b/mqtt-rpc/internal/update/roots.go @@ -0,0 +1,55 @@ +package update + +import ( + "crypto/tls" + "crypto/x509" + "net/http" + + "golang.org/x/crypto/x509roots/fallback/bundle" +) + +// Update downloads trust the reader firmware's CA bundle PLUS the Mozilla roots +// embedded in this binary. The firmware bundle is never refreshed; once it no +// longer validates GitHub, a reader trusting only it could never fetch another +// release, and only a site visit would fix it. Embedded roots ride every update, +// so each update refreshes them. +// +// A union, not a replacement: TLS here buys availability, not integrity — the +// Ed25519 signature and sha256 are the whole check of what gets installed — so +// any root that still works is worth keeping. Scoped to the download client: +// the broker connection keeps its own trust configuration. + +// rootsWith returns system plus every embedded root, honouring the bundle's +// distrust-after constraints. system is not modified. +func rootsWith(system *x509.CertPool) *x509.CertPool { + pool := system.Clone() + for r := range bundle.Roots() { + c, err := x509.ParseCertificate(r.Certificate) + if err != nil { + continue // one unparseable root must not take the rest down + } + if r.Constraint == nil { + pool.AddCert(c) + } else { + pool.AddCertWithConstraint(c, r.Constraint) + } + } + return pool +} + +// updateRoots is rootsWith the system pool, or the embedded roots alone when +// the system pool cannot be loaded. +func updateRoots() *x509.CertPool { + sys, err := x509.SystemCertPool() + if err != nil || sys == nil { + sys = x509.NewCertPool() + } + return rootsWith(sys) +} + +// downloadTransport is http.DefaultTransport's settings with updateRoots. +func downloadTransport() *http.Transport { + t := http.DefaultTransport.(*http.Transport).Clone() + t.TLSClientConfig = &tls.Config{RootCAs: updateRoots(), MinVersion: tls.VersionTLS12} + return t +} diff --git a/mqtt-rpc/internal/update/roots_test.go b/mqtt-rpc/internal/update/roots_test.go new file mode 100644 index 00000000..b57cf349 --- /dev/null +++ b/mqtt-rpc/internal/update/roots_test.go @@ -0,0 +1,61 @@ +package update + +import ( + "crypto/tls" + "crypto/x509" + "net/http" + "net/http/httptest" + "testing" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "golang.org/x/crypto/x509roots/fallback/bundle" +) + +func firstBundleRoot(t *testing.T) *x509.Certificate { + t.Helper() + for r := range bundle.Roots() { + if r.Constraint != nil { + continue + } + c, err := x509.ParseCertificate(r.Certificate) + require.NoError(t, err) + return c + } + t.Fatal("the embedded bundle has no unconstrained root") + return nil +} + +// The reader firmware's CA bundle goes stale and is never updated. Embedded +// roots ride every update, so a reader whose own bundle no longer validates +// GitHub can still fetch the next release. +func TestRootsWith_TrustsTheEmbeddedBundleWithAnEmptySystemPool(t *testing.T) { + root := firstBundleRoot(t) + _, err := root.Verify(x509.VerifyOptions{Roots: rootsWith(x509.NewCertPool())}) + assert.NoError(t, err) +} + +// The system pool is kept, not replaced: a root the firmware trusts that the +// bundle lacks must keep working. +func TestRootsWith_KeepsTheSystemPool(t *testing.T) { + srv := httptest.NewTLSServer(http.NotFoundHandler()) + t.Cleanup(srv.Close) + srvCert := srv.Certificate() // self-signed, so its own root + sys := x509.NewCertPool() + sys.AddCert(srvCert) + _, err := srvCert.Verify(x509.VerifyOptions{Roots: rootsWith(sys)}) + assert.NoError(t, err) +} + +func TestNew_DownloadsTrustTheUpdateRoots(t *testing.T) { + u := New(Layout{Dir: t.TempDir()}, "v1", func() {}, zerolog.Nop()) + tr, ok := u.HTTP.Transport.(*http.Transport) + require.True(t, ok, "the download client carries its own transport") + require.NotNil(t, tr.TLSClientConfig) + require.NotNil(t, tr.TLSClientConfig.RootCAs) + assert.GreaterOrEqual(t, tr.TLSClientConfig.MinVersion, uint16(tls.VersionTLS12)) + _, err := firstBundleRoot(t).Verify(x509.VerifyOptions{Roots: tr.TLSClientConfig.RootCAs}) + assert.NoError(t, err) + assert.NotNil(t, tr.Proxy, "proxy from environment, as http.DefaultTransport") +} diff --git a/mqtt-rpc/internal/update/updater.go b/mqtt-rpc/internal/update/updater.go index daa6ccdc..054b97a8 100644 --- a/mqtt-rpc/internal/update/updater.go +++ b/mqtt-rpc/internal/update/updater.go @@ -23,8 +23,15 @@ const ( defaultMaxBytes = 64 << 20 // the armv7 binary is ~10 MB downloadTimeout = 10 * time.Minute keepVersions = 3 + // staleDownloadAge: a temp file untouched this long belongs to no live + // download — ours is bounded by downloadTimeout, the guard's curl by 300 s. + staleDownloadAge = downloadTimeout + 5*time.Minute ) +// ErrAlreadyCommitted: this process already committed its update. A reply that +// raced the committing one gets this, and it is not a failure. +var ErrAlreadyCommitted = errors.New("update: already committed") + // Updater is the daemon side of a remote update. It never executes the binary it // downloads: the signature and hash are the whole check, and systemd starts it. type Updater struct { @@ -43,6 +50,7 @@ type Updater struct { busy bool attempt *readerrpc.UpdateInfo // this process's own attempt, if any probation bool + committed bool // this process ended its own probation } // New builds an Updater with production defaults. @@ -51,7 +59,7 @@ func New(l Layout, running string, restart func(), log zerolog.Logger) *Updater Layout: l, Keys: TrustedKeys, Running: running, - HTTP: &http.Client{Timeout: downloadTimeout}, + HTTP: &http.Client{Timeout: downloadTimeout, Transport: downloadTransport()}, Restart: restart, Log: log.With().Str("component", "update").Logger(), MaxBytes: defaultMaxBytes, @@ -65,8 +73,12 @@ func refuse(format string, a ...any) error { } // CheckPending runs once at startup: it decides whether this process is an update -// on probation (see the package-level CheckPending). +// on probation (see the package-level CheckPending), and clears temp files left +// by a download that a kill or power cut interrupted. func (u *Updater) CheckPending() error { + if err := CleanStale(u.Layout, staleDownloadAge); err != nil { + u.Log.Warn().Err(err).Msg("clean abandoned downloads") + } on, err := CheckPending(u.Layout, u.Running) u.mu.Lock() u.probation = on @@ -94,6 +106,12 @@ func (u *Updater) Preflight() error { if _, err := os.Stat(u.GuardTimerUnit); err != nil { return fmt.Errorf("rollback guard timer %s is not installed (re-run install.sh)", u.GuardTimerUnit) } + // Installed is not enough: a disabled timer never ticks, so a build that + // never reaches the broker would never be rolled back. `systemctl enable` + // records itself as this link, which needs no systemctl call to read. + if _, err := os.Stat(guardTimerWants(u.GuardTimerUnit)); err != nil { + return fmt.Errorf("rollback guard timer %s is not enabled (re-run install.sh)", u.GuardTimerUnit) + } active, err := ActiveVersion(u.Layout) if err != nil { return err @@ -166,6 +184,9 @@ func (u *Updater) run(p readerrpc.DaemonUpdateParams) { u.setState(readerrpc.UpdateStateFailed, msg) } + if err := CleanStale(u.Layout, staleDownloadAge); err != nil { + u.Log.Warn().Err(err).Msg("clean abandoned downloads") + } if err := u.download(p.URL, tmp, p.SHA256); err != nil { fail(err) return @@ -174,6 +195,12 @@ func (u *Updater) run(p readerrpc.DaemonUpdateParams) { fail(err) return } + // The rename must be durable before pending names it: otherwise a power cut + // could leave a guard armed for a binary whose directory entry was lost. + if err := syncDir(u.Layout.VersionsDir()); err != nil { + fail(err) + return + } u.setState(readerrpc.UpdateStateInstalling, "") // Arm the guard BEFORE swapping: a crash between the two leaves pending @@ -250,10 +277,14 @@ func (u *Updater) download(rawURL, dst, wantSHA string) error { } // Commit ends this build's probation, after checking that its own updater is -// usable. A commit that loses the race to the guard returns ErrNotPending. +// usable. A commit that loses the race to the guard returns ErrNotPending; one +// that loses to this process's own earlier commit returns ErrAlreadyCommitted. func (u *Updater) Commit() error { u.mu.Lock() defer u.mu.Unlock() + if u.committed { + return ErrAlreadyCommitted + } if !u.probation { return ErrNotPending } @@ -267,6 +298,7 @@ func (u *Updater) Commit() error { return err } u.probation = false + u.committed = true return nil } diff --git a/mqtt-rpc/internal/update/updater_test.go b/mqtt-rpc/internal/update/updater_test.go index 3f5e2b7b..0535b636 100644 --- a/mqtt-rpc/internal/update/updater_test.go +++ b/mqtt-rpc/internal/update/updater_test.go @@ -39,6 +39,8 @@ func newFixture(t *testing.T) *updaterFixture { require.NoError(t, os.WriteFile(f.l.Guard(), []byte("#!/bin/sh\n"), 0o755)) timer := filepath.Join(f.l.Dir, "mqtt-rpcd-guard.timer") require.NoError(t, os.WriteFile(timer, []byte("[Timer]\n"), 0o644)) + require.NoError(t, os.MkdirAll(filepath.Dir(guardTimerWants(timer)), 0o755)) + require.NoError(t, os.Symlink(timer, guardTimerWants(timer))) mux := http.NewServeMux() mux.HandleFunc("/mqtt-rpcd", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write(f.body) })