Fix web OTA upload and isolate OTA sessions across firmware and goTool.

Split ESP-NOW into core/master/slave modules, block non-OTA UART traffic during updates, and hold the host serial port exclusively so dashboard polling cannot interleave with firmware uploads.

Co-authored-by: Cursor <cursoragent@cursor.com>

goTool/api_serve.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -146,7 +147,11 @@ func serveOTAUpload(w http.ResponseWriter, r *http.Request, link *managedSerial,
 		if hub != nil {
 			hub.broadcastRaw(OTAProgress{Type: "ota_progress", Phase: "error", Message: err.Error()})
 		}
-		writeJSON(w, http.StatusServiceUnavailable, otaAPIResponse{Error: err.Error()})
+		status := http.StatusServiceUnavailable
+		if errors.Is(err, errOTAInProgress) {
+			status = http.StatusConflict
+		}
+		writeJSON(w, status, otaAPIResponse{Error: err.Error()})
 		return
 	}
 	writeJSON(w, http.StatusOK, otaAPIResponse{

goTool/cmd_ota.go

@@ -16,8 +16,7 @@ func runOTA(sp *serialPort, args []string) error {
 
 	sp.mu.Lock()
 	defer sp.mu.Unlock()
-	m := &managedSerial{quiet: false, sp: sp}
-	return runOTAOnPortUnlocked(m, data, func(p OTAProgress) {
+	return runOTAOnPortUnlocked(sp, data, func(p OTAProgress) {
 		switch p.Phase {
 		case "preparing", "ready":
 			fmt.Println(p.Message)

goTool/dashboard.go

@@ -101,7 +101,7 @@ func (h *wsHub) setState(st DashboardState) {
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -112,7 +112,7 @@ func (h *wsHub) register(c *websocket.Conn) {
 	h.mu.Unlock()
 
 	if data, err := json.Marshal(snap); err == nil {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -122,6 +122,30 @@ func (h *wsHub) unregister(c *websocket.Conn) {
 	h.mu.Unlock()
 }
 
+// writeJSON sends to one client; removes it on error or panic (closed connection).
+func (h *wsHub) writeJSON(c *websocket.Conn, data []byte) {
+	defer func() {
+		if recover() != nil {
+			h.unregister(c)
+		}
+	}()
+	if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
+		h.unregister(c)
+	}
+}
+
+func (h *wsHub) broadcastJSON(data []byte) {
+	h.mu.RLock()
+	conns := make([]*websocket.Conn, 0, len(h.clients))
+	for c := range h.clients {
+		conns = append(conns, c)
+	}
+	h.mu.RUnlock()
+	for _, c := range conns {
+		h.writeJSON(c, data)
+	}
+}
+
 func applyAccelSamples(clients []ClientView, samples []*pb.AccelSample) []ClientView {
 	if len(samples) == 0 {
 		return clients
@@ -338,7 +362,7 @@ func (h *wsHub) patchClientAccelStream(clientID uint32, enabled bool) {
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -397,7 +421,7 @@ func (h *wsHub) patchLiveStream(enabled bool) {
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -430,7 +454,7 @@ func (h *wsHub) patchClientTapNotify(clientID uint32, single, doubleTap, triple
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -455,7 +479,7 @@ func (h *wsHub) mergeAccel(samples []*pb.AccelSample) {
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -479,25 +503,16 @@ func (h *wsHub) mergeTap(events []*pb.TapEvent) {
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
 func (h *wsHub) broadcastRaw(v any) {
-	h.mu.RLock()
-	conns := make([]*websocket.Conn, 0, len(h.clients))
-	for c := range h.clients {
-		conns = append(conns, c)
-	}
-	h.mu.RUnlock()
-
 	data, err := json.Marshal(v)
 	if err != nil {
 		return
 	}
-	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
-	}
+	h.broadcastJSON(data)
 }
 
 func pollDashboard(link *managedSerial, portName string, last *DashboardState, streamCtl *accelStreamCtl, tapCtl *tapNotifyCtl) DashboardState {
@@ -575,6 +590,9 @@ func pollDashboard(link *managedSerial, portName string, last *DashboardState, s
 
 func applyBatteryToState(link *managedSerial, st *DashboardState) {
 	bat, err := link.BatteryStatusPoll(&pb.BatteryStatusRequest{AllClients: true})
+	if errors.Is(err, errUARTBusy) {
+		return
+	}
 	if err != nil {
 		log.Printf("battery poll: %v", err)
 		return
@@ -602,7 +620,7 @@ func (h *wsHub) mergeBattery(samples []batterySampleJSON) {
 		return
 	}
 	for _, c := range conns {
-		_ = c.WriteMessage(websocket.TextMessage, data)
+		h.writeJSON(c, data)
 	}
 }
 
@@ -619,7 +637,7 @@ func runBatteryPoller(link *managedSerial, hub *wsHub, interval time.Duration, s
 				continue
 			}
 			bat, err := link.BatteryStatusPoll(&pb.BatteryStatusRequest{AllClients: true})
-			if err != nil {
+			if errors.Is(err, errUARTBusy) || err != nil {
 				continue
 			}
 			hub.mergeBattery(batterySamplesFromPB(bat.GetSamples()))

goTool/ota_upload.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"fmt"
-	"log"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -69,16 +68,45 @@ const (
 )
 
 func runOTAUpload(m *managedSerial, firmware []byte, onProgress otaProgressFn) error {
+	push := func(phase, msg string) {
+		if onProgress == nil {
+			return
+		}
+		onProgress(OTAProgress{
+			Type: "ota_progress", Phase: phase, Step: otaStepMaster,
+			Percent: 0, Message: msg, MasterMessage: msg,
+		})
+	}
+	push("preparing", "UART wird vorbereitet…")
+
+	// Block until the UART is free, then hold m.mu for the entire upload so
+	// dashboard/API polling cannot interleave on the serial port.
 	m.mu.Lock()
-	defer m.mu.Unlock()
-	err := runOTAOnPortUnlocked(m, firmware, onProgress)
+	if m.otaActive {
+		m.mu.Unlock()
+		return errOTAInProgress
+	}
+	m.otaActive = true
+	if m.sp == nil {
+		if err := m.openLocked(); err != nil {
+			m.otaActive = false
+			m.mu.Unlock()
+			push("error", err.Error())
+			return err
+		}
+	}
+	sp := m.sp
+
+	err := runOTAOnPortUnlocked(sp, firmware, onProgress)
 	if err != nil {
 		m.invalidateLocked(err)
 	}
+	m.otaActive = false
+	m.mu.Unlock()
 	return err
 }
 
-func runOTAOnPortUnlocked(m *managedSerial, firmware []byte, onProgress otaProgressFn) error {
+func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgressFn) error {
 	if len(firmware) == 0 {
 		return fmt.Errorf("empty firmware")
 	}
@@ -120,32 +148,31 @@ func runOTAOnPortUnlocked(m *managedSerial, firmware []byte, onProgress otaProgr
 		onProgress(p)
 	}
 
-	if m.sp == nil {
-		if err := m.openLocked(); err != nil {
-			notify("error", "", 0, err.Error())
-			return err
-		}
-	}
-
-	sp := m.sp
-	if err := sp.port.SetReadTimeout(otaPrepareTimeout); err != nil {
+	if err := sp.port.SetReadTimeout(readTimeout); err != nil {
 		notify("error", "", 0, err.Error())
 		return err
 	}
-	defer sp.port.SetReadTimeout(readTimeout)
 
 	notify("preparing", otaStepMaster, 0, fmt.Sprintf("Master: OTA start (%d bytes)…", imageSize))
 
+	flushSerialInput(sp)
+
 	if err := writeUartMessage(sp, &pb.UartMessage{
 		Type: pb.MessageType_OTA_START,
 		Payload: &pb.UartMessage_OtaStart{
 			OtaStart: &pb.OtaStartPayload{TotalSize: uint32(imageSize)},
 		},
-	}, false); err != nil {
+	}); err != nil {
 		notify("error", "", 0, err.Error())
 		return err
 	}
 
+	if err := sp.port.SetReadTimeout(otaPrepareTimeout); err != nil {
+		notify("error", "", 0, err.Error())
+		return err
+	}
+	defer func() { _ = sp.port.SetReadTimeout(readTimeout) }()
+
 	ready, err := waitOtaStatus(sp, otaStReady, otaPrepareTimeout, func(msg string) {
 		notify("preparing", otaStepMaster, 2, msg)
 	})
@@ -179,7 +206,7 @@ func runOTAOnPortUnlocked(m *managedSerial, firmware []byte, onProgress otaProgr
 				Payload: &pb.UartMessage_OtaPayload{
 					OtaPayload: &pb.OtaPayload{Seq: seq, Data: chunk},
 				},
-			}, false); err != nil {
+			}); err != nil {
 				notify("error", "", 0, err.Error())
 				return err
 			}
@@ -219,7 +246,7 @@ func runOTAOnPortUnlocked(m *managedSerial, firmware []byte, onProgress otaProgr
 		Payload: &pb.UartMessage_OtaEnd{
 			OtaEnd: &pb.OtaEndPayload{},
 		},
-	}, false); err != nil {
+	}); err != nil {
 		notify("error", "", 0, err.Error())
 		return err
 	}
@@ -333,7 +360,7 @@ func queryOtaSlaveProgressLocked(sp *serialPort, clientID uint32,
 			},
 		},
 	}
-	if err := writeUartMessage(sp, req, false); err != nil {
+	if err := writeUartMessage(sp, req); err != nil {
 		return nil, err
 	}
 	if queryTimeout <= 0 {
@@ -487,14 +514,11 @@ func waitOtaComplete(sp *serialPort, timeout time.Duration,
 	}
 }
 
-func writeUartMessage(sp *serialPort, msg *pb.UartMessage, logFrame bool) error {
+func writeUartMessage(sp *serialPort, msg *pb.UartMessage) error {
 	frame, err := encodeUartMessage(msg)
 	if err != nil {
 		return err
 	}
-	if logFrame {
-		log.Printf("sending %s (%d frame bytes)", msg.Type, len(frame))
-	}
 	_, err = sp.port.Write(frame)
 	return err
 }
@@ -505,12 +529,24 @@ func waitOtaStatus(sp *serialPort, want uint32, timeout time.Duration, onPrepari
 		if time.Now().After(deadline) {
 			return nil, fmt.Errorf("timeout waiting for OTA status %d", want)
 		}
-		if err := sp.port.SetReadTimeout(time.Until(deadline)); err != nil {
+		readWait := time.Until(deadline)
+		if readWait > otaStatusPollTimeout {
+			readWait = otaStatusPollTimeout
+		}
+		if err := sp.port.SetReadTimeout(readWait); err != nil {
 			return nil, err
 		}
-		st, err := readOtaStatus(sp)
+		payload, err := uartframe.ReadFrame(sp.port, nil)
 		if err != nil {
-			return nil, err
+			continue
+		}
+		msg, err := decodeUartPayload(payload)
+		if err != nil || msg.GetType() != pb.MessageType_OTA_STATUS {
+			continue
+		}
+		st := msg.GetOtaStatus()
+		if st == nil {
+			continue
 		}
 		switch st.GetStatus() {
 		case want:
@@ -553,6 +589,22 @@ func encodeUartMessage(msg *pb.UartMessage) ([]byte, error) {
 	return uartframe.EncodeFrame(payload)
 }
 
+// flushSerialInput drops stale RX bytes (not full frames — avoids ReadFrame blocking).
+func flushSerialInput(sp *serialPort) {
+	if sp == nil {
+		return
+	}
+	_ = sp.port.SetReadTimeout(10 * time.Millisecond)
+	buf := make([]byte, 256)
+	deadline := time.Now().Add(50 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		n, err := sp.port.Read(buf)
+		if n == 0 || err != nil {
+			return
+		}
+	}
+}
+
 func decodeUartPayload(payload []byte) (*pb.UartMessage, error) {
 	if len(payload) == 0 {
 		return nil, fmt.Errorf("empty response")

goTool/serial_link.go

@@ -11,14 +11,18 @@ import (
 // errUARTBusy is returned when the port is held for OTA (poller should not treat as unplug).
 var errUARTBusy = errors.New("uart busy (OTA in progress)")
 
+// errOTAInProgress is returned when a second OTA upload is attempted while one is running.
+var errOTAInProgress = errors.New("OTA upload already in progress")
+
 // managedSerial keeps the UART open and reconnects after I/O failures or unplug.
 type managedSerial struct {
 	portName string
 	baud     int
 	quiet    bool
 
-	mu sync.Mutex
-	sp *serialPort
+	mu        sync.Mutex
+	sp        *serialPort
+	otaActive bool // UART held for firmware upload; poll/API must not interleave
 }
 
 func newManagedSerial(portName string, baud int) *managedSerial {
@@ -76,20 +80,17 @@ func (m *managedSerial) withPort(fn func(*serialPort) error) error {
 	return m.withPortLocked(false, fn)
 }
 
-// withPortPoll is like withPort but returns errUARTBusy instead of blocking during OTA.
+// withPortPoll is like withPort but returns errUARTBusy during OTA (no TryLock race).
 func (m *managedSerial) withPortPoll(fn func(*serialPort) error) error {
 	return m.withPortLocked(true, fn)
 }
 
-func (m *managedSerial) withPortLocked(try bool, fn func(*serialPort) error) error {
-	if try {
-		if !m.mu.TryLock() {
-			return errUARTBusy
-		}
-	} else {
-		m.mu.Lock()
-	}
+func (m *managedSerial) withPortLocked(poll bool, fn func(*serialPort) error) error {
+	m.mu.Lock()
 	defer m.mu.Unlock()
+	if m.otaActive {
+		return errUARTBusy
+	}
 
 	if m.sp == nil {
 		if err := m.openLocked(); err != nil {

goTool/uart/frame.go

@@ -9,7 +9,8 @@ import (
 const (
 	StartMarker = 0xAA
 	StopMarker  = 0xCC
-	MaxPayload  = 252
+	// Must match main/uart.h MAX_PAYLOAD_SIZE (MAX_BUF_SIZE - 4).
+	MaxPayload = 248
 )
 
 var (

goTool/webui/index.html

@@ -938,8 +938,21 @@
           return rows;
         },
         applyOTAProgress(p) {
-          this.ota.phase = p.phase || '';
-          this.ota.step = p.step || this.ota.step || '';
+          const prevPhase = this.ota.phase;
+          const prevStep = this.ota.step;
+          if (p.phase) {
+            // Ignore out-of-order master upload updates after distribution started.
+            if (!(p.phase === 'uploading' && prevPhase === 'distributing')) {
+              this.ota.phase = p.phase;
+            }
+          }
+          if (p.step) {
+            if (!(p.step === 'master' && (prevStep === 'slaves' || prevPhase === 'distributing'))) {
+              this.ota.step = p.step;
+            }
+          } else if (!this.ota.step) {
+            this.ota.step = '';
+          }
           this.ota.percent = p.percent ?? this.ota.percent;
           this.ota.message = p.message || '';
           if (p.image_size) this.ota.imageSize = p.image_size;