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

virtfs: refactor BlockOnce to accept (readFn, signalFn) 

BlockOnce now takes a value-reader and a signal source. The framework
handles the blocking loop: read → compare hash to base → if different
return → else wait on signal or timeout.

On timeout (ctx.Done), returns empty (not error) so clients re-open
cleanly.

BlockOnceRaw retained for queue-style handlers (bypass/pending,
proc/wait) that manage their own blocking.

EventValue adapter wraps a <-chan Event into BlockOnce-compatible
(readFn, signalFn) pair via a thin goroutine.
This commit is contained in:
Levi Neely 2026-08-13 20:44:50 +02:00
parent b5cb8c397d
commit 794673a5d0
6 changed files with 96 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmd/olliesrv/internal/agent/agent.go
View File

@ -237,6 +237,14 @@ func (ag *Agent) notifyChange() {
ag.signalMu.Unlock()
}
// SignalCh returns the current signal channel (closed on any change).
func (ag *Agent) SignalCh() <-chan struct{} {
ag.signalMu.Lock()
ch := ag.signalCh
ag.signalMu.Unlock()
return ch
}
// WaitChange blocks until the agent's state differs from current.
// Returns the new value and true, or ("", false) if ctx is cancelled.
func (ag *Agent) WaitChange(ctx context.Context, field, current string) (string, bool) {

44
cmd/olliesrv/internal/fs/spec.go
View File

@ -48,6 +48,9 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
treeSpecUID = u.Username
}
// Event subscription for eventwait (lives for server lifetime).
eventRead, eventSignal := session.EventValue(session.SubscribeEvents(context.Background(), "*"))
return virtfs.DirNode("/",
virtfs.UID(treeSpecUID),
virtfs.GID(treeSpecGID),
@ -131,11 +134,10 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
virtfs.FileNode("eventwait", 0444,
virtfs.Doc("Blocking read for server events"),
virtfs.Read(func() ([]byte, error) {
return nil, fmt.Errorf("eventwait: use blocking read")
}),
virtfs.BlockOnce(func(cctx context.Context, _ string) ([]byte, string, error) {
return session.WaitEvent(cctx)
data, _, _ := eventRead()
return data, nil
}),
virtfs.BlockOnce(eventRead, eventSignal),
),
virtfs.FileNode("generate", 0666,
virtfs.Doc("One-shot LLM generation"),
@ -405,20 +407,9 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl
}
return nil
}),
virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) {
merged, cancel := mergeCtx(s.Ctx, cctx)
defer cancel()
if base == "" {
base = a.Feed.Hash()
}
_, ok := a.WaitChange(merged, agent.WatchFeed, base)
if !ok {
return nil, base, merged.Err()
}
data := a.Feed.Get()
hash := a.Feed.Hash()
return data, hash, nil
}),
virtfs.BlockOnce(func() ([]byte, string, error) {
return a.Feed.Get(), a.Feed.Hash(), nil
}, a.SignalCh),
),
virtfs.FileNode("chat", 0444,
virtfs.StatOverride(chatStat),
@ -463,19 +454,10 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl
virtfs.Read(func() ([]byte, error) {
return []byte(a.State() + "\n"), nil
}),
virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) {
merged, cancel := mergeCtx(s.Ctx, cctx)
defer cancel()
if base == "" {
base = a.State()
}
v, ok := a.WaitChange(merged, agent.WatchState, base)
if !ok {
st := a.State()
return []byte(st + "\n"), st, nil
}
return []byte(v + "\n"), v, nil
}),
virtfs.BlockOnce(func() ([]byte, string, error) {
st := a.State()
return []byte(st + "\n"), st, nil
}, a.SignalCh),
),
virtfs.FileNode("log", 0444,
virtfs.Read(func() ([]byte, error) {

67
cmd/olliesrv/internal/session/event.go
View File

@ -2,7 +2,10 @@ package session
import (
"context"
"crypto/sha256"
"encoding/hex"
"strings"
"sync"
"github.com/simonfxr/pubsub"
)
@ -16,19 +19,6 @@ type Event struct {
}
// PublishEvent publishes an event to a specific topic.
// Topic hierarchy:
//
// session.{sid}.new/kill/pause/resume/rename
// session.{sid}.agent.{aid}.new/kill/state
//
// Wildcard routing publishes to all ancestor wildcards:
//
// session.{sid}.agent.{aid}.state -> also publishes to:
// session.{sid}.agent.{aid}.*
// session.{sid}.agent.*
// session.{sid}.*
// session.*
// *
func PublishEvent(topic, payload string) {
ev := Event{Topic: topic, Payload: payload}
eventBus.Publish(topic, ev)
@ -43,7 +33,6 @@ func PublishEvent(topic, payload string) {
}
// SubscribeEvents returns a channel that receives all events matching the topic.
// Use "*" for all events, "session.*" for all session events, etc.
func SubscribeEvents(ctx context.Context, topic string) <-chan Event {
ch := make(chan Event, 64)
sub := eventBus.SubscribeChan(topic, ch, pubsub.KeepOpen)
@ -55,20 +44,42 @@ func SubscribeEvents(ctx context.Context, topic string) <-chan Event {
return ch
}
// WaitEvent blocks until an event is available. Returns the event as "topic payload\n".
func WaitEvent(ctx context.Context) ([]byte, string, error) {
ch := SubscribeEvents(ctx, "*")
select {
case ev, ok := <-ch:
if !ok {
return nil, "", nil
// EventValue adapts an event channel into BlockOnce-compatible (readFn, signalFn).
// A goroutine drains the channel; each arrival updates the value and fires the signal.
func EventValue(ch <-chan Event) (readFn func() ([]byte, string, error), signalFn func() <-chan struct{}) {
var mu sync.Mutex
var data []byte
var hash string
sig := make(chan struct{})
go func() {
for ev := range ch {
line := ev.Topic
if ev.Payload != "" {
line += " " + ev.Payload
}
d := []byte(line + "\n")
h := sha256.Sum256(d)
mu.Lock()
data = d
hash = hex.EncodeToString(h[:8])
close(sig)
sig = make(chan struct{})
mu.Unlock()
}
line := ev.Topic
if ev.Payload != "" {
line += " " + ev.Payload
}
return []byte(line + "\n"), "", nil
case <-ctx.Done():
return nil, "", nil
}()
readFn = func() ([]byte, string, error) {
mu.Lock()
defer mu.Unlock()
out := make([]byte, len(data))
copy(out, data)
return out, hash, nil
}
signalFn = func() <-chan struct{} {
mu.Lock()
defer mu.Unlock()
return sig
}
return
}

4
cmd/toolsrv/internal/server/server.go
View File

@ -134,7 +134,7 @@ 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(ctx context.Context, _ string) ([]byte, string, error) {
virtfs.BlockOnceRaw(func(ctx context.Context, _ string) ([]byte, string, error) {
req := bypass.NextPending(ctx)
if req == nil {
return nil, "", fmt.Errorf("context cancelled")
@ -211,7 +211,7 @@ func Spec(srv *Server) virtfs.FsNodeDecl {
),
virtfs.FileNode("wait", 0444,
virtfs.Doc("Block until process exits, returns exit code"),
virtfs.BlockOnce(func(_ context.Context, _ string) ([]byte, string, error) {
virtfs.BlockOnceRaw(func(_ context.Context, _ string) ([]byte, string, error) {
exitCode := p.Wait()
return []byte(fmt.Sprintf("%d\n", exitCode)), "", nil
}),

33
virtfs/decl.go
View File

@ -127,8 +127,37 @@ func Stream(fn func(context.Context, string) ([]byte, string, error)) NodeOption
return func(d *FsNodeDecl) { d.Stream = fn }
}
// BlockOnce sets a blocking read handler that returns once.
func BlockOnce(fn func(context.Context, string) ([]byte, string, error)) NodeOption {
// BlockOnce sets a blocking read handler that returns once per open.
// readFn returns (data, hash, error) — the current value.
// signal returns a channel that is closed when the value may have changed.
// The framework blocks until readFn returns a hash different from base.
func BlockOnce(readFn func() ([]byte, string, error), signal func() <-chan struct{}) NodeOption {
return func(d *FsNodeDecl) {
d.BlockOnce = func(ctx context.Context, base string) ([]byte, string, error) {
for {
data, hash, err := readFn()
if err != nil {
return nil, base, err
}
if base == "" {
base = hash
}
if hash != base {
return data, hash, nil
}
select {
case <-signal():
case <-ctx.Done():
return nil, "", nil // timeout: empty response, client re-opens
}
}
}
}
}
// BlockOnceRaw sets a blocking read handler with custom blocking logic.
// Use for queue-style consumers where there is no "current value" to compare.
func BlockOnceRaw(fn func(context.Context, string) ([]byte, string, error)) NodeOption {
return func(d *FsNodeDecl) { d.BlockOnce = fn }
}

6
virtfs/decl_test.go
View File

@ -23,7 +23,7 @@ func TestNodeOptions(t *testing.T) {
})
t.Run("BlockOnce", func(t *testing.T) {
spec := FileNode("test", 0444, BlockOnce(func(ctx context.Context, base string) ([]byte, string, error) {
spec := FileNode("test", 0444, BlockOnceRaw(func(ctx context.Context, base string) ([]byte, string, error) {
return nil, "", nil
}))
if spec.BlockOnce == nil {
@ -249,7 +249,7 @@ func TestRenameDirNotFound(t *testing.T) {
func TestOpenLeafBlockOnce(t *testing.T) {
spec := DirNode("/",
FileNode("blocking", 0444, BlockOnce(func(ctx context.Context, base string) ([]byte, string, error) {
FileNode("blocking", 0444, BlockOnceRaw(func(ctx context.Context, base string) ([]byte, string, error) {
return []byte("data"), "next", nil
})),
)
@ -483,7 +483,7 @@ func TestValidateMultipleBlockingModes(t *testing.T) {
spec := DirNode("/",
FileNode("bad", 0444,
BlockOnce(func(context.Context, string) ([]byte, string, error) { return nil, "", nil }),
BlockOnceRaw(func(context.Context, string) ([]byte, string, error) { return nil, "", nil }),
Stream(func(context.Context, string) ([]byte, string, error) { return nil, "", nil }),
),
)