diff --git a/cmd/olliesrv/internal/bypass/broker.go b/cmd/olliesrv/internal/bypass/broker.go index 2f133ef..1725ca2 100644 --- a/cmd/olliesrv/internal/bypass/broker.go +++ b/cmd/olliesrv/internal/bypass/broker.go @@ -1,6 +1,7 @@ package bypass import ( + "context" "fmt" "sync" "time" @@ -84,8 +85,14 @@ func (b *Broker) Pending() []*Request { } // NextPending blocks until a bypass request is available and returns it. -func (b *Broker) NextPending() *Request { - return <-b.pendingCh +// Returns nil if context is cancelled before a request arrives. +func (b *Broker) NextPending(ctx context.Context) *Request { + select { + case req := <-b.pendingCh: + return req + case <-ctx.Done(): + return nil + } } // PendingByID returns a specific pending request. diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index 8387f12..380f82e 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -171,13 +171,13 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl { virtfs.Doc("Bypass broker for sandboxed command approval"), virtfs.FileNode("pending", 0444, virtfs.Doc("Blocks until bypass request; returns JSON {id, cmd, cwd, env}"), - virtfs.BlockOnce(func(_ context.Context, _ string) ([]byte, string, error) { + virtfs.BlockOnce(func(ctx context.Context, _ string) ([]byte, string, error) { if broker == nil { return nil, "", errNoBypass } - req := broker.NextPending() + req := broker.NextPending(ctx) if req == nil { - return nil, "", fmt.Errorf("no pending request") + return nil, "", fmt.Errorf("context cancelled") } data, err := json.Marshal(req) if err != nil { diff --git a/cmd/toolsrv/internal/bypass/bypass.go b/cmd/toolsrv/internal/bypass/bypass.go index 9dcc0e3..edba4fc 100644 --- a/cmd/toolsrv/internal/bypass/bypass.go +++ b/cmd/toolsrv/internal/bypass/bypass.go @@ -75,8 +75,14 @@ func Submit(ctx context.Context, cmd, cwd string, env map[string]string) (bool, // NextPending blocks until a request is available and returns it. // Used by the 9P pending file to expose requests to the approver. -func NextPending() *Request { - return <-pending +// Returns nil if context is cancelled before a request arrives. +func NextPending(ctx context.Context) *Request { + select { + case req := <-pending: + return req + case <-ctx.Done(): + return nil + } } // Resolve completes a pending request with the given decision. diff --git a/cmd/toolsrv/internal/server/server.go b/cmd/toolsrv/internal/server/server.go index f99c167..f03a275 100644 --- a/cmd/toolsrv/internal/server/server.go +++ b/cmd/toolsrv/internal/server/server.go @@ -134,8 +134,11 @@ func Spec(srv *Server) virtfs.FsNodeDecl { virtfs.DirNode("bypass", virtfs.FileNode("pending", 0444, virtfs.Doc("Blocks until bypass request; returns JSON {id, cmd, cwd, env}"), - virtfs.BlockOnce(func(_ context.Context, _ string) ([]byte, string, error) { - req := bypass.NextPending() + virtfs.BlockOnce(func(ctx context.Context, _ string) ([]byte, string, error) { + req := bypass.NextPending(ctx) + if req == nil { + return nil, "", fmt.Errorf("context cancelled") + } data, err := json.Marshal(req) if err != nil { return nil, "", err diff --git a/cmd/toolsrv/p9.go b/cmd/toolsrv/p9.go index b7c616e..5cf7535 100644 --- a/cmd/toolsrv/p9.go +++ b/cmd/toolsrv/p9.go @@ -24,6 +24,7 @@ type fid struct { entry virtfs.File readBuf []byte // buffer for read data readDone bool // true if file has been read + waitBase string // base string for blocking/stream reads dirCache []byte } @@ -375,7 +376,52 @@ func handleRead(cs *connState, fc *plan9.Fcall, tree *virtfs.Tree) *plan9.Fcall return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Data: nil} } - // Buffer read data if not already done + // Handle BlockOnce mode: block until content is available, then cache it + if f.entry.BlockingReadMode() { + // Serve from cache if we already have content (multi-Tread paging) + if f.readBuf != nil { + if int(fc.Offset) >= len(f.readBuf) { + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0} + } + return readSlice(fc, f.readBuf) + } + // First Tread: block for content + content, nextBase, err := f.entry.BlockingRead(cs.ctx, f.waitBase) + if err != nil { + return errFcall(fc, err.Error()) + } + if nextBase != "" { + cs.mu.Lock() + if ff, ok := cs.fids[fc.Fid]; ok { + ff.waitBase = nextBase + } + cs.mu.Unlock() + } + f.readBuf = content + return readSlice(fc, content) + } + + // Handle Stream mode: block indefinitely, stream data incrementally + if f.entry.StreamMode() { + content, nextBase, err := f.entry.BlockingRead(cs.ctx, f.waitBase) + if err != nil { + return errFcall(fc, err.Error()) + } + if nextBase != "" { + cs.mu.Lock() + if ff, ok := cs.fids[fc.Fid]; ok { + ff.waitBase = nextBase + } + cs.mu.Unlock() + } + count := int(fc.Count) + if count > len(content) { + count = len(content) + } + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(count), Data: content[:count]} + } + + // Buffer read data if not already done (normal non-blocking read) if f.readBuf == nil { data, err := f.entry.Read() if err != nil && err != io.EOF { @@ -399,6 +445,20 @@ func handleRead(cs *connState, fc *plan9.Fcall, tree *virtfs.Tree) *plan9.Fcall return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Data: f.readBuf[offset:end]} } +// readSlice returns a slice of content based on offset/count from fc. +func readSlice(fc *plan9.Fcall, content []byte) *plan9.Fcall { + var data []byte + off := int(fc.Offset) + if off < len(content) { + end := off + int(fc.Count) + if end > len(content) { + end = len(content) + } + data = content[off:end] + } + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(data)), Data: data} +} + func handleDirRead(cs *connState, f *fid, fc *plan9.Fcall, tree *virtfs.Tree) *plan9.Fcall { if fc.Offset == 0 || f.dirCache == nil { path := f.path