lkn
/
anvillm
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

p9: buffer Twrite chunks, dispatch on Tclunk 

The 9P client splits writes larger than msize into multiple Twrite
messages with increasing offsets. The server was dispatching each
Twrite immediately, so only the first chunk was ever processed —
causing JSON parse failures for payloads exceeding msize.

Fix: accumulate all Twrite data into a per-fid writeBuf, then dispatch
the fully-assembled payload in Tclunk. Tclunk can return Rerror if
dispatch fails, preserving error reporting to the caller.

This unblocks large bead descriptions/comments and message bodies
without any msize negotiation changes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Levi Neely 2026-03-13 23:17:57 +01:00
parent f3c23bc61e
commit 179a59bf16
1 changed files with 36 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
internal/p9/server.go
View File

@ -145,6 +145,10 @@ type fid struct {
path string
mode uint8
offset int64
// writeBuf accumulates Twrite chunks for a single logical write operation.
// The 9P client splits writes larger than msize into multiple Twrite messages
// with increasing offsets; we reassemble them here and process on Tclunk.
writeBuf []byte
// For streaming /events endpoint
eventCh <-chan *eventbus.Event
eventUnsub func()
@ -352,13 +356,20 @@ func (s *Server) handle(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
return s.stat(cs, fc)
case plan9.Tclunk:
cs.mu.Lock()
if f, ok := cs.fids[fc.Fid]; ok && f.eventUnsub != nil {
f, hasFid := cs.fids[fc.Fid]
if hasFid && f.eventUnsub != nil {
// Cancel event subscription before dropping the fid.
f.eventUnsub()
f.eventUnsub = nil
}
delete(cs.fids, fc.Fid)
cs.mu.Unlock()
// If the fid has buffered write data, dispatch it now.
if hasFid && len(f.writeBuf) > 0 {
if rfc := s.dispatchWrite(cs, f, fc.Tag); rfc.Type == plan9.Rerror {
return rfc
}
}
return &plan9.Fcall{Type: plan9.Rclunk, Tag: fc.Tag}
default:
return errFcall(fc, "not supported")
@ -688,16 +699,32 @@ func (s *Server) read(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
}
func (s *Server) write(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
cs.mu.RLock()
cs.mu.Lock()
f, ok := cs.fids[fc.Fid]
if !ok {
cs.mu.RUnlock()
cs.mu.Unlock()
return errFcall(fc, "bad fid")
}
path := f.path
cs.mu.RUnlock()
// Accumulate data into the per-fid write buffer at the given offset.
// The 9P client splits writes larger than msize into multiple Twrite messages
// with increasing offsets; we reassemble here and dispatch on Tclunk.
end := int(fc.Offset) + len(fc.Data)
if end > len(f.writeBuf) {
grown := make([]byte, end)
copy(grown, f.writeBuf)
f.writeBuf = grown
}
copy(f.writeBuf[fc.Offset:], fc.Data)
cs.mu.Unlock()
return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))}
}
input := strings.TrimSpace(string(fc.Data))
// dispatchWrite processes the fully-assembled write payload for a given path.
// Called from Tclunk after all Twrite chunks have been accumulated.
func (s *Server) dispatchWrite(cs *connState, f *fid, tag uint16) *plan9.Fcall {
fc := &plan9.Fcall{Tag: tag, Data: f.writeBuf}
path := f.path
input := strings.TrimSpace(string(f.writeBuf))
parts := strings.Split(strings.TrimPrefix(path, "/"), "/")
// Handle beads writes
@ -706,10 +733,10 @@ func (s *Server) write(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
cs.mu.Lock()
sessionID := cs.sessionID
cs.mu.Unlock()
if err := s.beads.Write(beadsPath, fc.Data, sessionID); err != nil {
if err := s.beads.Write(beadsPath, f.writeBuf, sessionID); err != nil {
return errFcall(fc, err.Error())
}
return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))}
return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(f.writeBuf))}
}
return errFcall(fc, "beads not initialized")
}
@ -985,7 +1012,7 @@ func (s *Server) write(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
cs.mu.Unlock()
// Parse JSON message
msg, err := mailbox.FromJSON(fc.Data)
msg, err := mailbox.FromJSON(f.writeBuf)
if err != nil {
return errFcall(fc, fmt.Sprintf("invalid message JSON: %v", err))
}