server: fix eventwait to use Stream instead of BlockOnce
EventValue was storing only the latest event and using hash comparison, which caused events to be overwritten if they arrived faster than the client could read them. Now eventwait uses Stream mode with EventStream which delivers each event as it arrives. Events won't be lost due to rapid arrival.
This commit is contained in:
parent
70c00541df
commit
adb08fc51d
|
|
@ -117,7 +117,7 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
|
|||
}
|
||||
|
||||
// Event subscription for eventwait (lives for server lifetime).
|
||||
eventRead, eventSignal := session.EventValue(session.SubscribeEvents(cfg.Ctx, "*"))
|
||||
eventRead, eventSignal := session.EventStream(session.SubscribeEvents(cfg.Ctx, "*"))
|
||||
|
||||
return virtfs.DirNode("/",
|
||||
virtfs.UID(treeSpecUID),
|
||||
|
|
@ -240,12 +240,8 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
|
|||
}),
|
||||
),
|
||||
virtfs.FileNode("eventwait", 0444,
|
||||
virtfs.Doc("Blocking read for server events"),
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
data, _, _ := eventRead()
|
||||
return data, nil
|
||||
}),
|
||||
virtfs.BlockOnce(eventRead, eventSignal),
|
||||
virtfs.Doc("Streaming read for server events"),
|
||||
virtfs.Stream(eventRead, eventSignal),
|
||||
),
|
||||
virtfs.FileNode("generate", 0666,
|
||||
virtfs.Doc("One-shot LLM generation"),
|
||||
|
|
|
|||
|
|
@ -2,8 +2,6 @@ package session
|
|||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
|
|
@ -44,12 +42,12 @@ func SubscribeEvents(ctx context.Context, topic string) <-chan Event {
|
|||
return ch
|
||||
}
|
||||
|
||||
// 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{}) {
|
||||
// EventStream returns functions suitable for virtfs.Stream - delivers events as they arrive.
|
||||
// Each read returns the next event; blocks until one is available.
|
||||
func EventStream(ch <-chan Event) (readFn func(base string) ([]byte, string, error), signalFn func() <-chan struct{}) {
|
||||
var mu sync.Mutex
|
||||
var data []byte
|
||||
var hash string
|
||||
var pending []byte
|
||||
var counter int64
|
||||
sig := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
|
|
@ -59,22 +57,24 @@ func EventValue(ch <-chan Event) (readFn func() ([]byte, string, error), signalF
|
|||
line += " " + ev.Payload
|
||||
}
|
||||
d := []byte(line + "\n")
|
||||
h := sha256.Sum256(d)
|
||||
mu.Lock()
|
||||
data = d
|
||||
hash = hex.EncodeToString(h[:8])
|
||||
pending = d
|
||||
counter++
|
||||
close(sig)
|
||||
sig = make(chan struct{})
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
|
||||
readFn = func() ([]byte, string, error) {
|
||||
readFn = func(base string) ([]byte, string, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
out := make([]byte, len(data))
|
||||
copy(out, data)
|
||||
return out, hash, nil
|
||||
if len(pending) == 0 {
|
||||
return nil, base, nil // no data, will block on signal
|
||||
}
|
||||
out := pending
|
||||
pending = nil
|
||||
return out, "", nil
|
||||
}
|
||||
signalFn = func() <-chan struct{} {
|
||||
mu.Lock()
|
||||
|
|
|
|||
|
|
@ -252,7 +252,6 @@ void Ollie9pClient::handleEvent(const QString &eventLine)
|
|||
refreshSessions();
|
||||
} else if (rest.startsWith(QLatin1String("bypass."))) {
|
||||
// Bypass event: payload is "id\tcmd\tcwd"
|
||||
qDebug() << "handleEvent: bypass event" << eventLine;
|
||||
if (payload.isEmpty()) return;
|
||||
const int tab1 = payload.indexOf(QLatin1Char('\t'));
|
||||
if (tab1 < 0) return;
|
||||
|
|
@ -262,7 +261,6 @@ void Ollie9pClient::handleEvent(const QString &eventLine)
|
|||
const QString id = payload.left(tab1).toString();
|
||||
const QString cmd = payload.mid(tab1 + 1, tab2 - tab1 - 1).toString();
|
||||
const QString cwd = payload.mid(tab2 + 1).toString();
|
||||
qDebug() << "handleEvent: emitting bypassRequested" << sessionId << id << cmd.left(50);
|
||||
emit bypassRequested(sessionId.toString(), id, cmd, cwd);
|
||||
}
|
||||
}
|
||||
|
|
@ -856,7 +854,6 @@ void Ollie9pClient::checkPendingBypass()
|
|||
QString cwd = obj["cwd"].toString();
|
||||
|
||||
if (!id.isEmpty() && !cmd.isEmpty()) {
|
||||
qDebug() << "checkPendingBypass: found pending" << id << cmd.left(50);
|
||||
emit bypassRequested(m_activeSessionId, id, cmd, cwd);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue