fix(rpc): harden mobile JSI bridge against reader races and stream desync

.gitignore

@@ -96,3 +96,6 @@ CLAUDE.md
 docs
 .codegraph
 .mcp.json
+
+# Scratch plans (not part of the codebase)
+plans/2*.md

go/bind/keybase.go

@@ -58,9 +58,22 @@ var (
 var (
 	jsReadyOnce sync.Once
 	jsReadyCh   = make(chan struct{})
-	connMutex   sync.Mutex // Protects conn operations
+	connMutex   sync.Mutex // Protects conn and connEpoch
 )
 
+// connEpoch increments every time ensureConnection successfully (re)dials.
+// Callers that observed a failure on a particular conn capture the epoch
+// alongside it (while holding connMutex) and pass it to ResetIfCurrent,
+// which only tears down the connection if nothing has re-dialed since —
+// otherwise some other caller already recovered and this call is a stale,
+// no-op complaint. Protected by connMutex.
+var connEpoch int64
+
+// lastReadEpoch is the epoch of the connection the most recent ReadArr call
+// read (or is about to read) from, captured atomically with the connection
+// reference itself. See LastReadEpoch. Protected by connMutex.
+var lastReadEpoch int64
+
 func describeConn(c net.Conn) string {
 	if c == nil {
 		return "<nil>"
@@ -571,6 +584,7 @@ func WriteArr(b []byte) (err error) {
 		}
 	}
 	currentConn := conn
+	currentEpoch := connEpoch
 	connMutex.Unlock()
 
 	if currentConn == nil {
@@ -586,7 +600,17 @@ func WriteArr(b []byte) (err error) {
 	if n != len(bytes) {
 		log("Go: WriteArr short write conn=%s wrote=%d expected=%d appState=%s",
 			describeConn(currentConn), n, len(bytes), appStateForLog())
-		return errors.New("Did not write all the data")
+		// Not reachable through LoopbackConn today, whose Write is
+		// all-or-nothing. If a future transport can short-write, the peer's
+		// framer is left holding a partial frame and every later write would
+		// be consumed as its remainder, so drop the connection rather than
+		// corrupt the stream indefinitely. ResetIfCurrent (rather than Reset)
+		// so this doesn't tear down a connection some concurrent caller has
+		// already redialed since currentConn was captured above.
+		if ierr := ResetIfCurrent(currentEpoch); ierr != nil {
+			log("failed to Reset after short write: %v", ierr)
+		}
+		return fmt.Errorf("Did not write all the data: wrote %d of %d", n, len(bytes))
 	}
 	return nil
 }
@@ -595,7 +619,12 @@ func WriteArr(b []byte) (err error) {
 var buffer []byte
 
 // ReadArr is a blocking read for msgpack rpc data.
-// It is called serially by the mobile run loops.
+// It must still be called serially by the mobile run loops: conn.Read
+// ordering depends on it, and the msgpack::unpacker behind onDataFromGo
+// on the C++ side is not thread-safe. The returned slice is this call's own
+// copy, which only protects the caller of ReadArr — a second concurrent
+// caller would still interleave its Read into the shared package-level
+// buffer above, garbling frame content, not just ordering.
 func ReadArr() (data []byte, err error) {
 	defer func() { err = flattenError(err) }()
 
@@ -611,6 +640,8 @@ func ReadArr() (data []byte, err error) {
 		}
 	}
 	currentConn := conn
+	currentEpoch := connEpoch
+	lastReadEpoch = currentEpoch
 	connMutex.Unlock()
 
 	if currentConn == nil {
@@ -633,15 +664,21 @@ func ReadArr() (data []byte, err error) {
 		}
 		// Deliver data even if err != nil (allowed by the net.Conn
 		// contract); a persistent failure is returned by a later call.
-		// Returning a view of the shared buffer is safe because gomobile
-		// copies the bytes across the boundary and ReadArr is called
-		// serially.
-		return buffer[0:n], nil
+		//
+		// Copy out of the shared buffer rather than returning a view of it:
+		// gomobile copies again at the JNI/ObjC boundary regardless, so this
+		// costs one extra memcpy of a few KB on a call already crossing a
+		// language boundary, and it removes buffer aliasing as a hazard.
+		out := make([]byte, n)
+		copy(out, buffer[:n])
+		return out, nil
 	}
 
 	if err != nil {
-		// Attempt to fix the connection
-		if ierr := Reset(); ierr != nil {
+		// Attempt to fix the connection. ResetIfCurrent so a stale failure
+		// on an already-superseded conn doesn't tear down one a concurrent
+		// WriteArr just redialed.
+		if ierr := ResetIfCurrent(currentEpoch); ierr != nil {
 			log("failed to Reset: %v", ierr)
 		}
 		return nil, fmt.Errorf("Read error: %s", err)
@@ -677,21 +714,98 @@ func ensureConnection() error {
 			log("ensureConnection: Dial failed after restart: %v", err)
 			return fmt.Errorf("failed to dial after loopback restart: %s", err)
 		}
-		log("ensureConnection: loopback server restarted successfully conn=%s appState=%s",
-			describeConn(conn), appStateForLog())
+		connEpoch++
+		log("ensureConnection: loopback server restarted successfully conn=%s epoch=%d appState=%s",
+			describeConn(conn), connEpoch, appStateForLog())
 		return nil
 	}
-	log("Go: Established loopback connection conn=%s appState=%s",
-		describeConn(conn), appStateForLog())
+	connEpoch++
+	log("Go: Established loopback connection conn=%s epoch=%d appState=%s",
+		describeConn(conn), connEpoch, appStateForLog())
 	return nil
 }
 
-// Reset resets the socket connection
+// Reset unconditionally resets the socket connection. Use this only when the
+// caller genuinely means "tear down whatever connection is current" (e.g.
+// iOS invalidate, Android destroy/engineReset) — it will happily close a
+// connection some concurrent failure-driven caller never saw fail. Callers
+// reacting to a failure on a specific connection should use ResetIfCurrent
+// instead so a stale complaint can't clobber a connection that has already
+// been recovered.
 func Reset() error {
 	connMutex.Lock()
 	defer connMutex.Unlock()
+	return resetLocked()
+}
+
+// LastReadEpoch returns the epoch of the connection the most recent ReadArr
+// call read its bytes from. A caller that is about to hand off a batch of
+// bytes it just got back from ReadArr (e.g. a platform reader loop, right
+// before passing the data into the native bridge for parsing) should call
+// this immediately after ReadArr returns and capture the result there, not
+// later once some asynchronous handler for that batch actually runs —
+// ensureConnection may have re-dialed by then, and passing the wrong
+// (newer) epoch to ResetIfCurrent would tear down a good connection
+// instead of leaving it alone.
+//
+// This is exact only because there is exactly one permanent ReadArr caller
+// for the life of the process (enforced on both platforms: a single serial
+// reader loop). With one reader, nothing can start a second ReadArr call
+// between this one's internal epoch capture and this accessor being read
+// afterward, so the value can never be stale by the time the caller reads
+// it. A design with concurrent readers would need to thread the epoch
+// through ReadArr's return value instead.
+func LastReadEpoch() int64 {
+	connMutex.Lock()
+	defer connMutex.Unlock()
+	return lastReadEpoch
+}
+
+// ResetIfCurrent closes the connection only if epoch (obtained implicitly by
+// whoever last dialed) still matches the live connection's epoch. If
+// ensureConnection has re-dialed since, the
+// epoch has moved on and this call is a stale no-op: some other caller
+// already recovered, and closing the newer connection would silently
+// discard whatever was just written to it.
+func ResetIfCurrent(epoch int64) error {
+	_, err := resetIfCurrent(epoch)
+	return err
+}
+
+// ResetIfCurrentDidReset behaves exactly like ResetIfCurrent, additionally
+// reporting whether the reset actually ran (epoch matched connEpoch) as
+// opposed to being a stale no-op (epoch was already superseded).
+//
+// Platform readers need this distinction: they also reset their local parser
+// state (resetRecv/nativeResetRecv) alongside the Go connection, and doing so
+// unconditionally after a stale no-op drops bytes already in flight on a
+// connection nothing here actually touched. See Kb.mm and KbModule.kt.
+func ResetIfCurrentDidReset(epoch int64) bool {
+	didReset, err := resetIfCurrent(epoch)
+	if err != nil {
+		log("Go: ResetIfCurrentDidReset: reset error: %v", err)
+	}
+	return didReset
+}
+
+// resetIfCurrent is the shared implementation behind ResetIfCurrent and
+// ResetIfCurrentDidReset. didReset reports whether epoch matched connEpoch
+// (i.e. whether resetLocked actually ran), independent of whether resetLocked
+// itself returned an error.
+func resetIfCurrent(epoch int64) (didReset bool, err error) {
+	connMutex.Lock()
+	defer connMutex.Unlock()
+	if epoch != connEpoch {
+		log("Go: ResetIfCurrent skipped, stale epoch=%d current=%d appState=%s",
+			epoch, connEpoch, appStateForLog())
+		return false, nil
+	}
+	return true, resetLocked()
+}
 
-	log("Go: Reset start conn=%s appState=%s", describeConn(conn), appStateForLog())
+// resetLocked does the actual teardown. Must be called with connMutex held.
+func resetLocked() error {
+	log("Go: Reset start conn=%s epoch=%d appState=%s", describeConn(conn), connEpoch, appStateForLog())
 	if conn != nil {
 		conn.Close()
 		conn = nil