diff --git a/Makefile b/Makefile index 464b474..2311d7c 100644 --- a/Makefile +++ b/Makefile @@ -125,3 +125,53 @@ fmt: ## Run go fmt against code. .PHONY: vet vet: ## Run go vet against code. go vet ./... + +# --- Hub deploy (RO rootfs: binaries live on /data) ----------------------- +# Hub runs Dropbear. Older images lack /usr/libexec/sftp-server; -O uses +# legacy scp. Harmless on images that ship openssh sftp-server. +# +# make deploy-hub +# make deploy-hub HUB=192.168.0.10 +# +# Installs linux/arm64 prod binaries into /data/opt/bigfred/bin (preferred +# over the image copies in /opt) and restarts microinit services. +HUB ?= 192.168.0.1 +HUB_USER ?= root +HUB_SSH ?= $(HUB_USER)@$(HUB) +SCP ?= scp +SCP_OPTS ?= -O +SSH ?= ssh +HUB_BIN_DIR ?= /data/opt/bigfred/bin + +.PHONY: hub-arm64 deploy-hub +hub-arm64: web-build + CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -tags prod -ldflags="-s -w $(VERSION_LDFLAGS)" \ + -o bin/loco-server-linux-arm64 ./pkgs/bigfred/server + CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -ldflags="$(VERSION_LDFLAGS)" \ + -o bin/bigfred-remote-icmp-linux-arm64 ./pkgs/bigfred/remote-icmp + @ls -lh bin/loco-server-linux-arm64 bin/bigfred-remote-icmp-linux-arm64 + +# Upload next to the target and rename: writing in place fails with ETXTBSY +# once the hub is running the /data copy, and rename(2) swaps the inode +# atomically. dcc-bus runs `bigfred dcc-bus …` from the same binary, so those +# daemons keep the old inode until they are restarted too. +deploy-hub: hub-arm64 + @test -f bin/loco-server-linux-arm64 || { echo "error: bin/loco-server-linux-arm64 missing" >&2; exit 1; } + @test -f bin/bigfred-remote-icmp-linux-arm64 || { echo "error: bin/bigfred-remote-icmp-linux-arm64 missing" >&2; exit 1; } + $(SSH) $(HUB_SSH) 'mkdir -p $(HUB_BIN_DIR)' + $(SCP) $(SCP_OPTS) bin/loco-server-linux-arm64 $(HUB_SSH):$(HUB_BIN_DIR)/.bigfred.new + $(SCP) $(SCP_OPTS) bin/bigfred-remote-icmp-linux-arm64 $(HUB_SSH):$(HUB_BIN_DIR)/.bigfred-remote-icmp.new + $(SSH) $(HUB_SSH) 'set -e; \ + cd $(HUB_BIN_DIR); \ + chmod 755 .bigfred.new .bigfred-remote-icmp.new; \ + mv -f .bigfred.new bigfred; \ + mv -f .bigfred-remote-icmp.new bigfred-remote-icmp; \ + command -v setcap >/dev/null && setcap cap_net_raw+ep bigfred-remote-icmp || true; \ + rc=0; \ + microinit restart bigfred || rc=1; \ + microinit restart remote-icmp || rc=1; \ + for p in $$(microinit list 2>/dev/null | grep -o "^dcc-bus-[^ ]*" || true); do \ + echo "restarting $$p"; \ + microinit restart "$$p" || rc=1; \ + done; \ + exit $$rc' diff --git a/pkgs/bigfred/dcc-bus/daemon.go b/pkgs/bigfred/dcc-bus/daemon.go index 1501a0a..c6355a1 100644 --- a/pkgs/bigfred/dcc-bus/daemon.go +++ b/pkgs/bigfred/dcc-bus/daemon.go @@ -366,7 +366,7 @@ func New(ctx context.Context, log *logrus.Logger, cfg Config) (*Daemon, error) { log.Info("dcc-bus ws metrics enabled") } - wsSrv := ws.NewServer(ws.ServerConfig{ + wsCfg := ws.ServerConfig{ Verifier: verifier, Hub: hub, Router: ws.NewRouterAdapter(router), @@ -386,7 +386,11 @@ func New(ctx context.Context, log *logrus.Logger, cfg Config) (*Daemon, error) { AllowedOrigins: cfg.AllowedOrigins, Verifier: verifier, }), - }) + } + if h, ok := st.(ws.StationHealth); ok { + wsCfg.StationHealth = h + } + wsSrv := ws.NewServer(wsCfg) srv := &http.Server{ Addr: net.JoinHostPort(cfg.BindAddr, strconv.Itoa(int(cfg.Port))), diff --git a/pkgs/bigfred/dcc-bus/ws/handler.go b/pkgs/bigfred/dcc-bus/ws/handler.go index 593e0d5..ca7c095 100644 --- a/pkgs/bigfred/dcc-bus/ws/handler.go +++ b/pkgs/bigfred/dcc-bus/ws/handler.go @@ -82,6 +82,16 @@ type Server struct { // the daemon binds to loopback because the reverse proxy on // loco-server already validates Origin). AllowedOrigins []string + + // stationHealth is optional; Z21 reports UDP reachability so + // /healthz can fail closed while the station is reconnecting. + stationHealth StationHealth +} + +// StationHealth is implemented by command-station drivers that can +// report whether the physical bus is reachable (Z21 serial heartbeat). +type StationHealth interface { + Reachable() bool } // ServerConfig collects the few knobs Server takes at construction. @@ -101,6 +111,8 @@ type ServerConfig struct { // ProgrammingEnabled opens the loco.cvRead / cvWrite / addrGet / // addrSet frames. Off by default. ProgrammingEnabled bool + // StationHealth, when set, is consulted by /healthz. + StationHealth StationHealth } // NewServer returns a ready-to-mount Server. Heartbeat and dead-man @@ -124,19 +136,20 @@ func NewServer(cfg ServerConfig) *Server { log = logrus.New() } return &Server{ - verifier: cfg.Verifier, - hub: cfg.Hub, - router: cfg.Router, - log: log, - layoutID: cfg.LayoutID, - csID: cfg.CommandStation, - speedSteps: steps, - heartbeatSecs: hb, - deadmanSecs: dms, - AllowedOrigins: cfg.AllowedOrigins, - metrics: cfg.Metrics, - slotsDiag: cfg.SlotsDiag, + verifier: cfg.Verifier, + hub: cfg.Hub, + router: cfg.Router, + log: log, + layoutID: cfg.LayoutID, + csID: cfg.CommandStation, + speedSteps: steps, + heartbeatSecs: hb, + deadmanSecs: dms, + AllowedOrigins: cfg.AllowedOrigins, + metrics: cfg.Metrics, + slotsDiag: cfg.SlotsDiag, programmingEnabled: cfg.ProgrammingEnabled, + stationHealth: cfg.StationHealth, } } @@ -160,13 +173,23 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) } case "/healthz": - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte(`{"status":"ok"}`)) + s.handleHealthz(w) default: http.NotFound(w, r) } } +func (s *Server) handleHealthz(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/json") + if s.stationHealth != nil && !s.stationHealth.Reachable() { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"status":"unhealthy","code":"station_unreachable"}`)) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ok"}`)) +} + // handleWS authenticates, upgrades, registers the session and runs // the read loop until ctx ends or the client disconnects. func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { @@ -250,8 +273,8 @@ func (s *Server) readLoop(ctx context.Context, sess *Session) { sess.Close(errors.WsCodeSessionReadLoopDone) } s.log.WithFields(logrus.Fields{ - "sessionId": sess.ID, - "userId": sess.UserID, + "sessionId": sess.ID, + "userId": sess.UserID, "userSessionsRemaining": len(s.hub.SessionsForUser(sess.UserID)), }).Info("dcc-bus session closed") // Give the browser time to reconnect before firing the dead-man's diff --git a/pkgs/bigfred/dcc-bus/ws/handler_healthz_test.go b/pkgs/bigfred/dcc-bus/ws/handler_healthz_test.go new file mode 100644 index 0000000..ad3dac1 --- /dev/null +++ b/pkgs/bigfred/dcc-bus/ws/handler_healthz_test.go @@ -0,0 +1,49 @@ +package ws + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +type fakeStationHealth struct{ ok bool } + +func (f fakeStationHealth) Reachable() bool { return f.ok } + +func TestHealthzOKWhenNoStationHealth(t *testing.T) { + s := NewServer(ServerConfig{}) + rec := httptest.NewRecorder() + s.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } +} + +func TestHealthzOKWhenStationReachable(t *testing.T) { + s := NewServer(ServerConfig{StationHealth: fakeStationHealth{ok: true}}) + rec := httptest.NewRecorder() + s.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } +} + +func TestHealthzUnavailableWhenStationUnreachable(t *testing.T) { + s := NewServer(ServerConfig{StationHealth: fakeStationHealth{ok: false}}) + rec := httptest.NewRecorder() + s.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want 503", rec.Code) + } + var body struct { + Status string `json:"status"` + Code string `json:"code"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("json: %v", err) + } + if body.Status != "unhealthy" || body.Code != "station_unreachable" { + t.Fatalf("body = %+v", body) + } +} diff --git a/pkgs/loco/commandstation/z21.go b/pkgs/loco/commandstation/z21.go index 6265815..2fd90fc 100644 --- a/pkgs/loco/commandstation/z21.go +++ b/pkgs/loco/commandstation/z21.go @@ -26,9 +26,16 @@ const ( z21BcAllLocos uint32 = 0x00010000 ) +const ( + z21ReconnectBackoff = 2 * time.Second + z21HeartbeatInterval = 10 * time.Second + z21HeartbeatTimeout = 2 * time.Second +) + // NewZ21Roco constructor func NewZ21Roco(netAddr string, netPort uint16) (*Z21Roco, error) { roco := Z21Roco{ + addr: fmt.Sprintf("%s:%d", netAddr, netPort), Timeout: time.Second * 10, ReadInfoTimeout: 1500 * time.Millisecond, wasPowerCutOff: false, @@ -37,11 +44,23 @@ func NewZ21Roco(netAddr string, netPort uint16) (*Z21Roco, error) { stop: make(chan struct{}), metrics: newZ21Metrics(), } - return &roco, roco.connect(fmt.Sprintf("%s:%d", netAddr, netPort)) + if err := roco.dial(); err != nil { + return nil, err + } + go roco.readLoop() + go roco.heartbeatLoop() + return &roco, nil } type Z21Roco struct { - conn net.Conn + addr string + conn net.Conn + connMu sync.RWMutex + // reachable is false while a reconnect is in progress or the last + // serial-number heartbeat failed. + reachable atomic.Bool + reconnecting atomic.Bool + // Timeout bounds CV programming-track read/verify cycles, which can // be slow on some decoders. Timeout time.Duration @@ -67,6 +86,10 @@ type Z21Roco struct { obsCh chan LocoObservation stop chan struct{} + // broadcastsWanted is set when ObserveStates has requested + // LAN_X_LOCO_INFO push; reconnect re-sends the flags because a + // new UDP socket is a new Z21 LAN session. + broadcastsWanted atomic.Bool // enableBroadcastsOnce lazily turns on LAN_X_LOCO_INFO push the // first time a consumer asks for observations. enableBroadcastsOnce sync.Once @@ -107,14 +130,29 @@ type fnState struct { B29_31 byte // DB8 } -func (z *Z21Roco) connect(netAddr string) error { - conn, err := net.Dial("udp", netAddr) +func (z *Z21Roco) currentConn() net.Conn { + z.connMu.RLock() + defer z.connMu.RUnlock() + return z.conn +} + +// Reachable reports whether the last UDP dial / serial heartbeat succeeded. +func (z *Z21Roco) Reachable() bool { + return z.reachable.Load() +} + +func (z *Z21Roco) dial() error { + conn, err := net.Dial("udp", z.addr) if err != nil { return fmt.Errorf("UDP dial error while connecting to Roco Z21: %s", err) } + z.connMu.Lock() + old := z.conn z.conn = conn - // initialize cache + channels (defensive: in case the struct was - // assembled without NewZ21Roco) + z.connMu.Unlock() + if old != nil { + _ = old.Close() + } z.fnStateMu.Lock() if z.fnStateCache == nil { z.fnStateCache = make(map[LocoAddr]fnState) @@ -132,8 +170,83 @@ func (z *Z21Roco) connect(netAddr string) error { if z.metrics == nil { z.metrics = newZ21Metrics() } - logrus.WithField("remote", netAddr).Info("z21 command station: UDP socket open") - go z.readLoop() + logrus.WithField("remote", z.addr).Info("z21 command station: UDP socket open") + return nil +} + +func (z *Z21Roco) doReconnect() { + if !z.reconnecting.CompareAndSwap(false, true) { + return + } + defer z.reconnecting.Store(false) + z.reachable.Store(false) + for { + select { + case <-z.stop: + return + default: + } + if err := z.dial(); err != nil { + logrus.WithError(err).WithField("remote", z.addr).Warn("z21 reconnect failed, retrying") + select { + case <-z.stop: + return + case <-time.After(z21ReconnectBackoff): + } + continue + } + z.restoreBroadcasts() + if err := z.pingSerial(); err != nil { + logrus.WithError(err).WithField("remote", z.addr).Warn("z21 reconnect probe failed, retrying") + select { + case <-z.stop: + return + case <-time.After(z21ReconnectBackoff): + } + continue + } + return + } +} + +func (z *Z21Roco) heartbeatLoop() { + // UDP Dial succeeds without a peer; probe immediately so /healthz + // does not stay green for a full interval against a dead Railbox. + if err := z.pingSerial(); err != nil { + logrus.WithError(err).WithField("remote", z.addr).Warn("z21 initial heartbeat failed, reconnecting") + z.doReconnect() + } + ticker := time.NewTicker(z21HeartbeatInterval) + defer ticker.Stop() + for { + select { + case <-z.stop: + return + case <-ticker.C: + if err := z.pingSerial(); err != nil { + logrus.WithError(err).WithField("remote", z.addr).Warn("z21 heartbeat failed, reconnecting") + z.doReconnect() + } + } + } +} + +func (z *Z21Roco) pingSerial() error { + z.ioMu.Lock() + defer z.ioMu.Unlock() + z.beginSync() + defer z.endSync() + if _, err := z.write(z21SerialNumberProbe); err != nil { + return err + } + _, err := z.awaitMatching(z21HeartbeatTimeout, func(pkt []byte) bool { + return len(pkt) >= 4 && binary.LittleEndian.Uint16(pkt[2:4]) == 0x0010 + }) + if err != nil { + z.reachable.Store(false) + return err + } + z.reachable.Store(true) return nil } @@ -162,7 +275,15 @@ func (z *Z21Roco) CleanUp() error { close(z.stop) } logrus.Info("z21 command station: closing UDP socket") - return z.conn.Close() + z.connMu.Lock() + conn := z.conn + z.conn = nil + z.connMu.Unlock() + z.reachable.Store(false) + if conn != nil { + return conn.Close() + } + return nil } func (Z *Z21Roco) markBuildTrackPowerOff() { @@ -174,6 +295,7 @@ func (Z *Z21Roco) markBuildTrackPowerOff() { // Z21 LAN_X_LOCO_INFO broadcast so the station pushes state changes — // including those made by external handsets — to this client. func (z *Z21Roco) ObserveStates() <-chan LocoObservation { + z.broadcastsWanted.Store(true) z.enableBroadcastsOnce.Do(func() { if err := z.enableLocoInfoBroadcast(); err != nil { logrus.WithError(err).Warn("z21: enabling loco-info broadcast failed; push may not work") @@ -184,6 +306,15 @@ func (z *Z21Roco) ObserveStates() <-chan LocoObservation { return z.obsCh } +func (z *Z21Roco) restoreBroadcasts() { + if !z.broadcastsWanted.Load() { + return + } + if err := z.enableLocoInfoBroadcast(); err != nil { + logrus.WithError(err).Warn("z21: re-enabling loco-info broadcast after reconnect failed") + } +} + // SubscribeLocoInfo implements LocoInfoSubscriber. It sends // LAN_X_GET_LOCO_INFO (§4.1) which subscribes addr for unsolicited // LAN_X_LOCO_INFO under broadcast flag 0x00000001 — the mechanism that @@ -223,8 +354,13 @@ func (z *Z21Roco) readLoop() { default: } - _ = z.conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) - n, err := z.conn.Read(buf) + conn := z.currentConn() + if conn == nil { + z.doReconnect() + continue + } + _ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + n, err := conn.Read(buf) if err != nil { if ne, ok := err.(net.Error); ok && ne.Timeout() { continue @@ -234,8 +370,8 @@ func (z *Z21Roco) readLoop() { return default: z.metrics.incr(&z.metrics.rxErrors) - logrus.Debugf("z21 read error: %v", err) - time.Sleep(100 * time.Millisecond) + logrus.WithError(err).Warn("z21 read error, reconnecting") + z.doReconnect() continue } } diff --git a/pkgs/loco/commandstation/z21_drive_decode_test.go b/pkgs/loco/commandstation/z21_drive_decode_test.go index 4fd30e5..b257200 100644 --- a/pkgs/loco/commandstation/z21_drive_decode_test.go +++ b/pkgs/loco/commandstation/z21_drive_decode_test.go @@ -248,3 +248,21 @@ func TestDecodeLocoDriveFromLocoInfo(t *testing.T) { }) } } + +func TestBuildProgReadPacketDataLen(t *testing.T) { + t.Parallel() + z := &Z21Roco{} + pkt := z.buildProgReadPacket(CV{Num: 8}) + if got := binary.LittleEndian.Uint16(pkt[0:2]); got != 0x0009 { + t.Fatalf("DataLen = %#04x, want 0x0009 (LAN_X_CV_READ §6.1)", got) + } + if len(pkt) != 9 { + t.Fatalf("len = %d, want 9", len(pkt)) + } + want := []byte{0x09, 0x00, 0x40, 0x00, 0x23, 0x11, 0x00, 0x07, 0x35} + for i := range want { + if pkt[i] != want[i] { + t.Fatalf("byte %d = %#02x, want %#02x (pkt=% X)", i, pkt[i], want[i], pkt) + } + } +} diff --git a/pkgs/loco/commandstation/z21_proto.go b/pkgs/loco/commandstation/z21_proto.go index 0818f26..46b5c27 100644 --- a/pkgs/loco/commandstation/z21_proto.go +++ b/pkgs/loco/commandstation/z21_proto.go @@ -2,6 +2,7 @@ package commandstation import ( "encoding/binary" + "errors" "github.com/sirupsen/logrus" ) @@ -61,7 +62,7 @@ func (z *Z21Roco) buildPomWriteByte(lcv LocoCV) []byte { // ===== PROG (Programming Track / Direct Mode) ===== // Read: LAN_X_CV_READ (23 11) func (z *Z21Roco) buildProgReadPacket(cv CV) []byte { - const dataLen, header = 0x000B, 0x0040 + const dataLen, header = 0x0009, 0x0040 cvWire := cv.Translate() x := []byte{0x23, 0x11, byte(cvWire >> 8), byte(cvWire & 0xFF)} @@ -120,7 +121,7 @@ func (z *Z21Roco) buildTrackPower(on bool) []byte { // SetTrackPower implements TrackPowerController via LAN_X_SET_TRACK_POWER_*. func (z *Z21Roco) SetTrackPower(on bool) error { - if z == nil || z.conn == nil { + if z == nil || z.currentConn() == nil { return ErrTrackPowerUnsupported } _, err := z.write(z.buildTrackPower(on)) @@ -226,10 +227,16 @@ func (z *Z21Roco) buildSetLocoSpeed(addr LocoAddr, speed uint8, forward bool, sp func (z *Z21Roco) write(b []byte) (n int, err error) { logrus.Debugf("write: % X", b) - n, err = z.conn.Write(b) + conn := z.currentConn() + if conn == nil { + z.metrics.incr(&z.metrics.txErrors) + return 0, errors.New("z21: not connected") + } + n, err = conn.Write(b) if err != nil { z.metrics.incr(&z.metrics.txErrors) logrus.WithError(err).Warn("z21 command station: UDP write failed") + go z.doReconnect() return n, err } for _, pkt := range splitZ21Datagram(b[:n]) {