From 15002a88d728749fb36f93d19eae669c7cc97a6c Mon Sep 17 00:00:00 2001 From: AMATH <116212274+amathxbt@users.noreply.github.com> Date: Sat, 4 Jul 2026 09:11:08 +0100 Subject: [PATCH] fix(plugin): serialize writes to the socket connection with a dedicated mutex ListenForInbound() spawns a new goroutine per inbound FSM message. Each goroutine can independently call sendProtoMsg() -> sendLengthPrefixed() -> conn.Write(), either as a direct response to the FSM (Genesis/Begin/Check/ Deliver/End) or via a nested StateRead/StateWrite call made from within the contract handler. The pending/requestContract maps are correctly guarded by the existing mutex (p.l), but the underlying net.Conn write path had no equivalent protection. Concurrent writers can interleave mid-message on the wire, corrupting the length-prefixed framing that both sides rely on to delimit messages -- this would manifest as garbled/undecodable protobuf payloads or a stuck connection under any real concurrency (e.g. a contract issuing a StateRead while another goroutine sends a response for a prior message). Fix: add Plugin.writeMu and hold it for the duration of the length-prefixed write in sendLengthPrefixed(), so at most one goroutine writes to the connection at a time. --- contract/plugin.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/contract/plugin.go b/contract/plugin.go index 9105256..9bc8ea0 100644 --- a/contract/plugin.go +++ b/contract/plugin.go @@ -24,7 +24,8 @@ type Plugin struct { conn net.Conn // the underlying unix sock file connection pending map[uint64]chan isFSMToPlugin_Payload // the outstanding requests from the contract requestContract map[uint64]*Contract // maps request IDs to their contract context for concurrent operations - l sync.Mutex // thread safety + l sync.Mutex // thread safety for pending/requestContract maps + writeMu sync.Mutex // thread safety for concurrent writes to the socket connection config Config // general app config } @@ -266,6 +267,12 @@ func (p *Plugin) sendLengthPrefixed(bz []byte) *PluginError { // create the length prefix (4 bytes, big endian) lengthPrefix := make([]byte, 4) binary.BigEndian.PutUint32(lengthPrefix, uint32(len(bz))) + // serialize writes: ListenForInbound() spawns a goroutine per inbound message, and each + // can independently write to the shared connection (either as a direct FSM response or + // via a nested StateRead/StateWrite call). Without this lock, concurrent writers can + // interleave mid-message and corrupt the length-prefixed framing on the wire. + p.writeMu.Lock() + defer p.writeMu.Unlock() // write the message (length prefixed) if _, er := p.conn.Write(append(lengthPrefix, bz...)); er != nil { return ErrFailedPluginWrite(er)