feat(zap): add error response protocol and concurrent request handling

Add MsgResponseFlag for error responses over the wire, enable concurrent
request dispatch with per-goroutine panic recovery, and simplify State
type to use canonical vm.State. Update consensus dependency to v1.22.67.

go.mod

@@ -32,6 +32,6 @@ require (
 	golang.org/x/sys v0.40.0 // indirect
 	golang.org/x/text v0.33.0 // indirect
 	gonum.org/v1/gonum v0.16.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20251222181119-0a764e51fe1b // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 )

go.sum

@@ -51,7 +51,7 @@ golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
 golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ=
 golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE=
 gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20251222181119-0a764e51fe1b h1:Mv8VFug0MP9e5vUxfBcE3vUkV6CImK3cMNMIDFjmzxU=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516 h1:sNrWoksmOyF5bvJUcnmbeAmQi8baNhqg5IWaI3llQqU=
 google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

zap/transport.go

@@ -218,9 +218,17 @@ func (c *Conn) readLoop() {
 		c.mu.Unlock()
 
 		if ok {
-			ch <- &response{
-				msgType: msgType,
-				payload: respPayload,
+			// Check for error response (MsgResponseFlag set)
+			if msgType&MsgResponseFlag != 0 {
+				errMsg := string(respPayload)
+				ch <- &response{
+					err: fmt.Errorf("remote error: %s", errMsg),
+				}
+			} else {
+				ch <- &response{
+					msgType: msgType,
+					payload: respPayload,
+				}
 			}
 		}
 	}
@@ -335,7 +343,7 @@ func (c *ServerConn) Write(reqID uint32, msgType MessageType, payload []byte) er
 		_ = c.conn.SetWriteDeadline(time.Now().Add(c.config.WriteTimeout))
 	}
 
-	if err := WriteMessage(c.writer, msgType|MsgResponseFlag, respBuf.Bytes()); err != nil {
+	if err := WriteMessage(c.writer, msgType, respBuf.Bytes()); err != nil {
 		return err
 	}
 	return c.writer.Flush()
@@ -441,21 +449,25 @@ func (s *Server) handleConn(ctx context.Context, conn *ServerConn) {
 			continue
 		}
 
-		// Handle request
-		respType, respPayload, err := s.handler.Handle(ctx, msgType, payload)
-		if err != nil {
-			// Send error response
-			errBuf := GetBuffer()
-			errBuf.WriteString(err.Error())
-			_ = conn.Write(reqID, msgType|MsgResponseFlag, errBuf.Bytes())
-			PutBuffer(errBuf)
-			continue
-		}
+		// Handle request concurrently to avoid blocking on long-running
+		// operations like WaitForEvent. Write is protected by writeMu.
+		go func(reqID uint32, msgType MessageType, payload []byte) {
+			defer func() {
+				if r := recover(); r != nil {
+					// Send panic as error response
+					errMsg := fmt.Sprintf("handler panic: %v", r)
+					conn.Write(reqID, msgType|MsgResponseFlag, []byte(errMsg))
+				}
+			}()
 
-		// Send response
-		if err := conn.Write(reqID, respType, respPayload); err != nil {
-			return
-		}
+			respType, respPayload, err := s.handler.Handle(ctx, msgType, payload)
+			if err != nil {
+				conn.Write(reqID, msgType|MsgResponseFlag, []byte(err.Error()))
+				return
+			}
+
+			conn.Write(reqID, respType, respPayload)
+		}(reqID, msgType, payload)
 	}
 }
 

zap/vm.go

@@ -3,16 +3,6 @@
 
 package zap
 
-// State represents the VM state
-type State uint8
-
-const (
-	StateUnspecified State = iota
-	StateStateSyncing
-	StateBootstrapping
-	StateNormalOp
-)
-
 // Error represents VM errors
 type Error uint8
 
@@ -233,20 +223,24 @@ func (m *InitializeResponse) Decode(r *Reader) error {
 	return err
 }
 
-// SetStateRequest contains state change request
+// SetStateRequest contains state change request.
+// State values are defined in github.com/luxfi/vm (vm.State):
+//
+//	Unknown=0, Starting=1, Syncing=2, Bootstrapping=3,
+//	Ready=4, Degraded=5, Stopping=6, Stopped=7
 type SetStateRequest struct {
-	State State
+	State uint8
 }
 
 // Encode serializes SetStateRequest to the buffer
 func (m *SetStateRequest) Encode(buf *Buffer) {
-	buf.WriteUint8(uint8(m.State))
+	buf.WriteUint8(m.State)
 }
 
 // Decode deserializes SetStateRequest from the reader
 func (m *SetStateRequest) Decode(r *Reader) error {
-	v, err := r.ReadUint8()
-	m.State = State(v)
+	var err error
+	m.State, err = r.ReadUint8()
 	return err
 }