Merge branch 'testing' of ssh://git.stormux.org:1101/storm/cthulhu into testing
This commit is contained in:
commit
5f7806eb32
14 files changed
+264
-4
No files matched your search
@@ -0,0 +1,48 @@
|
|||||||
|
# Tentacle diagnostics
|
||||||
|
|
||||||
|
With debug logging enabled, the reader records Tentacle activation, failure,
|
||||||
|
disable, process exit and cleanup; failed requests; remote approval and lease
|
||||||
|
opening; rejected remote operations; queue exhaustion; policy cancellation and
|
||||||
|
lease revocation. Records identify the installed Tentacle and process session.
|
||||||
|
Remote records also carry an epoch (connection generation). Revocation records
|
||||||
|
include the last operation sequence and milliseconds remaining until expiry.
|
||||||
|
A negative remaining duration indicates that the deadline has passed.
|
||||||
|
|
||||||
|
`remote keyboard capture ended` describes keyboard cleanup, not necessarily the
|
||||||
|
original disconnect. Look earlier for `tentacle remote revoked`, queue or policy
|
||||||
|
records, and `tentacle child diagnostic`. An `expired` revocation is the reader's
|
||||||
|
local heartbeat watchdog; it does not by itself prove internet failure. A child
|
||||||
|
`network`, `timeout` or `closed` reason describes its relay transport; `peer`
|
||||||
|
means membership/origin validation failed, and `protocol` means a message was
|
||||||
|
rejected. These are observations, not proof of which remote machine or network
|
||||||
|
component caused the problem. Child diagnostics are explicitly attributed claims.
|
||||||
|
|
||||||
|
Remote's updated executable emits version 1 JSON lines on stderr. Older readers
|
||||||
|
discard these. Both reader and Tentacle must be rebuilt to get relay diagnostics;
|
||||||
|
rebuilding only the reader adds host diagnostics. This does not change stdout
|
||||||
|
RPC or the protocol negotiation version.
|
||||||
|
|
||||||
|
A record has `cthulhu_diagnostic: 1`, numeric `epoch` and `elapsed_ms`, and fixed
|
||||||
|
`phase` and `reason` classifications. A failed reader call can include an
|
||||||
|
`operation` from the remote RPC vocabulary. Epoch zero means a process-level
|
||||||
|
record or call without an attached generation. Connection final elapsed time is
|
||||||
|
measured from the start of the connection attempt; reader-call elapsed time is
|
||||||
|
that call's duration. Other records use zero when duration is not measured.
|
||||||
|
|
||||||
|
The host validates the vocabulary in `internal/tentacle/diagnostics.go`, rejects
|
||||||
|
unknown fields and records over 1024 bytes, and retains the existing 64 KiB
|
||||||
|
lifetime stderr budget. Partial final lines and ordinary stderr are discarded.
|
||||||
|
Remote stops emitting after 48 KiB, with one final `diagnostics`/`limit` record,
|
||||||
|
so repeated connection attempts cannot exhaust the host limit through these
|
||||||
|
diagnostics alone. Restart the Tentacle to reset that budget.
|
||||||
|
It never inserts arbitrary stderr into logs. Diagnostic records contain no
|
||||||
|
channel keys, lease tokens, clipboard content or relay payloads. Existing speech
|
||||||
|
and other debug logging can contain private content; redact logs before sharing.
|
||||||
|
|
||||||
|
Focused automated checks cover streaming/chunk boundaries, oversized and invalid
|
||||||
|
records, lease lifecycle logging and token omission, and remote error categories.
|
||||||
|
For live validation, use a rebuilt reader and Remote executable, connect and
|
||||||
|
exchange speech/keyboard input, return with F11, then disconnect. Confirm the log
|
||||||
|
contains an opening/ready/closed sequence and a host lease revocation for the same
|
||||||
|
session and epoch. If an unexpected drop recurs, preserve the entire debug log
|
||||||
|
before restarting, including the records preceding keyboard cleanup.
|
||||||
@@ -72,6 +72,7 @@ func (runtime *tentacleRuntime) start_remote_capture(ctx context.Context, sessio
|
|||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
case <-leaseDone:
|
case <-leaseDone:
|
||||||
|
runtime.logger.Debug("remote capture lease ended", "session", session.Identity(), "epoch", lease.Epoch)
|
||||||
leaseDone = nil
|
leaseDone = nil
|
||||||
end()
|
end()
|
||||||
case request := <-requests:
|
case request := <-requests:
|
||||||
@@ -119,7 +120,7 @@ func (runtime *tentacleRuntime) start_remote_capture(ctx context.Context, sessio
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if event.Ended != "" {
|
if event.Ended != "" {
|
||||||
runtime.logger.Debug("remote keyboard capture ended", "reason", event.Ended)
|
runtime.logger.Debug("remote keyboard capture ended", "reason", event.Ended, "session", session.Identity(), "epoch", lease.Epoch)
|
||||||
_ = session.PublishRemoteFocus(lease, tentaclewire.RemoteFocusLocal, event.Ended)
|
_ = session.PublishRemoteFocus(lease, tentaclewire.RemoteFocusLocal, event.Ended)
|
||||||
runtime.report("Keyboard returned to this computer.")
|
runtime.report("Keyboard returned to this computer.")
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -433,6 +433,9 @@ func (runtime *tentacleRuntime) report(message string) {
|
|||||||
// makeManager runs on the lifecycle worker, never on input or GTK callbacks.
|
// makeManager runs on the lifecycle worker, never on input or GTK callbacks.
|
||||||
func (runtime *tentacleRuntime) makeManager() (*tentacle.Manager, error) {
|
func (runtime *tentacleRuntime) makeManager() (*tentacle.Manager, error) {
|
||||||
manager, err := runtime.make_manager()
|
manager, err := runtime.make_manager()
|
||||||
|
if err == nil {
|
||||||
|
manager.SetLogger(runtime.logger)
|
||||||
|
}
|
||||||
if err == nil && runtime.remotePolicy != nil {
|
if err == nil && runtime.remotePolicy != nil {
|
||||||
err = manager.UseRemotePolicy(runtime.remotePolicy)
|
err = manager.UseRemotePolicy(runtime.remotePolicy)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,112 @@
|
|||||||
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
|
||||||
|
package tentacle
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"log/slog"
|
||||||
|
"slices"
|
||||||
|
)
|
||||||
|
|
||||||
|
// SetLogger configures diagnostics for subsequently started sessions.
|
||||||
|
func (manager *Manager) SetLogger(logger *slog.Logger) {
|
||||||
|
manager.mutex.Lock()
|
||||||
|
defer manager.mutex.Unlock()
|
||||||
|
manager.logger = logger
|
||||||
|
}
|
||||||
|
|
||||||
|
func (session *Session) log(message string, args ...any) {
|
||||||
|
if session.logger != nil {
|
||||||
|
fields := []any{"tentacle", session.spec.ID, "session", session.identity}
|
||||||
|
session.logger.Debug(message, append(fields, args...)...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Error strings can contain paths or extension-controlled text. Classify them
|
||||||
|
// without copying that text into host diagnostics.
|
||||||
|
func diagnosticError(err error) string {
|
||||||
|
if err == nil {
|
||||||
|
return "none"
|
||||||
|
}
|
||||||
|
for _, item := range []struct {
|
||||||
|
err error
|
||||||
|
name string
|
||||||
|
}{
|
||||||
|
{context.DeadlineExceeded, "timeout"}, {context.Canceled, "cancelled"},
|
||||||
|
{ErrProtocol, "protocol"}, {ErrLimit, "limit"}, {ErrExited, "exited"},
|
||||||
|
{ErrDisabled, "disabled"}, {ErrPermission, "permission"}, {ErrBusy, "capacity"},
|
||||||
|
{ErrUnavailable, "unavailable"}, {ErrSupervision, "supervision"}, {ErrContainment, "containment"},
|
||||||
|
} {
|
||||||
|
if errors.Is(err, item.err) {
|
||||||
|
return item.name
|
||||||
|
}
|
||||||
|
}
|
||||||
|
var rpc *RPCError
|
||||||
|
if errors.As(err, &rpc) {
|
||||||
|
return "rpc"
|
||||||
|
}
|
||||||
|
return "other"
|
||||||
|
}
|
||||||
|
|
||||||
|
func diagnosticOperation(method string) string {
|
||||||
|
switch method {
|
||||||
|
case "remote.open", "remote.close", "remote.heartbeat", "remote.present", "remote.key", "remote.focus", "remote.braille":
|
||||||
|
return method
|
||||||
|
default:
|
||||||
|
return "other"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stderr remains untrusted. Only this bounded, versioned vocabulary is admitted;
|
||||||
|
// ordinary stderr, credentials, server text and malformed records are discarded.
|
||||||
|
// Fields are reported as child claims, never interpreted as host events.
|
||||||
|
type childDiagnostic struct {
|
||||||
|
Operation string `json:"operation,omitempty"`
|
||||||
|
Version int `json:"cthulhu_diagnostic"`
|
||||||
|
Epoch uint64 `json:"epoch"`
|
||||||
|
Phase string `json:"phase"`
|
||||||
|
Reason string `json:"reason"`
|
||||||
|
ElapsedMS int64 `json:"elapsed_ms"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type diagnosticWriter struct {
|
||||||
|
session *Session
|
||||||
|
line []byte
|
||||||
|
overflow bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (writer *diagnosticWriter) Write(data []byte) (int, error) {
|
||||||
|
for _, b := range data {
|
||||||
|
if b == '\n' {
|
||||||
|
if !writer.overflow {
|
||||||
|
writer.record()
|
||||||
|
}
|
||||||
|
writer.line = writer.line[:0]
|
||||||
|
writer.overflow = false
|
||||||
|
} else if len(writer.line) < 1024 {
|
||||||
|
writer.line = append(writer.line, b)
|
||||||
|
} else {
|
||||||
|
writer.overflow = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return len(data), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (writer *diagnosticWriter) record() {
|
||||||
|
var d childDiagnostic
|
||||||
|
if decode_strict(json.RawMessage(writer.line), &d) != nil || d.Version != 1 || d.ElapsedMS < 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if d.Operation != "" && diagnosticOperation(d.Operation) == "other" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !slices.Contains([]string{"diagnostics", "opening", "connecting", "waiting", "ready", "stopping", "closed", "failed", "relay-read", "relay-write", "relay-message", "reader-call", "keyboard", "event-stream"}, d.Phase) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !slices.Contains([]string{"none", "requested", "cancelled", "timeout", "closed", "network", "trust", "protocol", "limit", "peer", "lease", "keyboard", "state", "stale", "other"}, d.Reason) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writer.session.log("tentacle child diagnostic", "epoch", d.Epoch, "phase", d.Phase, "operation", d.Operation, "reason", d.Reason, "elapsed_ms", d.ElapsedMS)
|
||||||
|
}
|
||||||
@@ -0,0 +1,71 @@
|
|||||||
|
package tentacle
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"log/slog"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestDiagnosticStreamBoundsAndPrivacy(t *testing.T) {
|
||||||
|
var output bytes.Buffer
|
||||||
|
session := &Session{logger: slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})), identity: "fixture"}
|
||||||
|
writer := &diagnosticWriter{session: session}
|
||||||
|
good := `{"cthulhu_diagnostic":1,"epoch":3,"phase":"relay-read","reason":"timeout","elapsed_ms":42}` + "\n"
|
||||||
|
for _, bad := range []string{
|
||||||
|
"private stderr\n", strings.Repeat("x", 2048) + "\n",
|
||||||
|
strings.Replace(good, `"timeout"`, `"private secret"`, 1),
|
||||||
|
strings.Replace(good, `"elapsed_ms":42`, `"elapsed_ms":42,"secret":"private"`, 1),
|
||||||
|
} {
|
||||||
|
_, _ = writer.Write([]byte(bad))
|
||||||
|
}
|
||||||
|
if output.Len() != 0 {
|
||||||
|
t.Fatalf("untrusted text admitted: %s", output.String())
|
||||||
|
}
|
||||||
|
for _, b := range []byte(good) {
|
||||||
|
_, _ = writer.Write([]byte{b})
|
||||||
|
}
|
||||||
|
if strings.Count(output.String(), "tentacle child diagnostic") != 1 || !strings.Contains(output.String(), `"epoch":3`) {
|
||||||
|
t.Fatal(output.String())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRemoteLifecycleDiagnosticsOmitAuthorityTokens(t *testing.T) {
|
||||||
|
var output bytes.Buffer
|
||||||
|
manager := test_manager(t, DefaultLimits(), nil)
|
||||||
|
manager.SetLogger(slog.New(slog.NewJSONHandler(&output, &slog.HandlerOptions{Level: slog.LevelDebug})))
|
||||||
|
// Exercise the real lease lifecycle with a logger attached, without a child.
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
session := &Session{manager: manager, logger: manager.logger, identity: "fixture", state: Active, ctx: ctx, outputContext: ctx,
|
||||||
|
spec: Spec{SessionEvents: true, RemoteGrants: []string{"remote-present"}}, outbound: make(chan outgoing, 16)}
|
||||||
|
action, err := session.AuthorizeRemote("controller", []string{"remote-present"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
params := tentaclewire.RemoteOpen{Action: action.Action, Epoch: action.Epoch, Role: action.Role, Grants: action.Grants}
|
||||||
|
session.mutex.Lock()
|
||||||
|
raw, _ := json.Marshal(params)
|
||||||
|
value, failure := session.remote_call_locked("remote.open", raw)
|
||||||
|
session.mutex.Unlock()
|
||||||
|
if failure != nil {
|
||||||
|
t.Fatal(failure)
|
||||||
|
}
|
||||||
|
lease := value.(tentaclewire.RemoteLease)
|
||||||
|
session.RevokeRemote()
|
||||||
|
logs := output.String()
|
||||||
|
for _, want := range []string{"remote authorized", "remote lease opened", "remote revoked", `"reason":"local"`} {
|
||||||
|
if !strings.Contains(logs, want) {
|
||||||
|
t.Fatalf("missing %s: %s", want, logs)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, secret := range []string{action.Action, lease.Lease} {
|
||||||
|
if strings.Contains(logs, secret) {
|
||||||
|
t.Fatal("authority token logged")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"errors"
|
"errors"
|
||||||
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -53,6 +54,7 @@ type Stats struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type Manager struct {
|
type Manager struct {
|
||||||
|
logger *slog.Logger // Copied into sessions under mutex.
|
||||||
mutex sync.Mutex
|
mutex sync.Mutex
|
||||||
limits Limits
|
limits Limits
|
||||||
sessions map[string]*Session
|
sessions map[string]*Session
|
||||||
@@ -157,7 +159,7 @@ func (manager *Manager) Start(ctx context.Context, spec Spec) (*Session, error)
|
|||||||
}
|
}
|
||||||
sessionContext, cancel := context.WithCancel(ctx)
|
sessionContext, cancel := context.WithCancel(ctx)
|
||||||
outputContext, revokeOutput := context.WithCancel(sessionContext)
|
outputContext, revokeOutput := context.WithCancel(sessionContext)
|
||||||
session := &Session{manager: manager, spec: spec, identity: hex.EncodeToString(identity[:]),
|
session := &Session{logger: manager.logger, manager: manager, spec: spec, identity: hex.EncodeToString(identity[:]),
|
||||||
outputContext: outputContext, revokeOutput: revokeOutput,
|
outputContext: outputContext, revokeOutput: revokeOutput,
|
||||||
ctx: sessionContext, cancel: cancel, state: Starting, ready: make(chan struct{}), done: make(chan struct{}),
|
ctx: sessionContext, cancel: cancel, state: Starting, ready: make(chan struct{}), done: make(chan struct{}),
|
||||||
stop: make(chan struct{}), revoked: make(chan struct{}), outbound: make(chan outgoing, queueSize),
|
stop: make(chan struct{}), revoked: make(chan struct{}), outbound: make(chan outgoing, queueSize),
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func (session *Session) run() {
|
func (session *Session) run() {
|
||||||
|
session.log("tentacle starting", "supervised", len(session.manager.supervisorProgram) != 0, "contained", session.manager.containment != nil)
|
||||||
var cleanupError error
|
var cleanupError error
|
||||||
starting := true
|
starting := true
|
||||||
ready := func() {
|
ready := func() {
|
||||||
@@ -39,6 +40,7 @@ func (session *Session) run() {
|
|||||||
} else if session.state != Failed {
|
} else if session.state != Failed {
|
||||||
session.fail_locked(ErrExited)
|
session.fail_locked(ErrExited)
|
||||||
}
|
}
|
||||||
|
session.log("tentacle stopped", "state", session.state, "cleanup_reason", diagnosticError(cleanupError))
|
||||||
session.mutex.Unlock()
|
session.mutex.Unlock()
|
||||||
session.callbacks.Wait()
|
session.callbacks.Wait()
|
||||||
// A retained status/session handle must not retain unsent argument text.
|
// A retained status/session handle must not retain unsent argument text.
|
||||||
@@ -202,6 +204,7 @@ func (session *Session) run() {
|
|||||||
} else {
|
} else {
|
||||||
<-exited
|
<-exited
|
||||||
}
|
}
|
||||||
|
session.log("tentacle process reaped", "exit_code", command.ProcessState.ExitCode(), "success", command.ProcessState.Success())
|
||||||
workers.Wait()
|
workers.Wait()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
@@ -357,8 +360,9 @@ func (session *Session) read_replies(reader io.Reader) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (session *Session) drain_diagnostics(reader io.Reader) {
|
func (session *Session) drain_diagnostics(reader io.Reader) {
|
||||||
count, err := io.Copy(io.Discard, io.LimitReader(reader, MaxDiagnosticBytes+1))
|
count, err := io.Copy(&diagnosticWriter{session: session}, io.LimitReader(reader, MaxDiagnosticBytes+1))
|
||||||
if count > MaxDiagnosticBytes {
|
if count > MaxDiagnosticBytes {
|
||||||
|
session.log("tentacle diagnostic budget exceeded", "bytes", count)
|
||||||
session.fail(ErrLimit)
|
session.fail(ErrLimit)
|
||||||
} else if err != nil && session.ctx.Err() == nil {
|
} else if err != nil && session.ctx.Err() == nil {
|
||||||
session.fail(ErrExited)
|
session.fail(ErrExited)
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ const (
|
|||||||
MaxReaderArgumentBytes = 48 * 1024
|
MaxReaderArgumentBytes = 48 * 1024
|
||||||
MaxArgumentBytes = 16 * 1024
|
MaxArgumentBytes = 16 * 1024
|
||||||
MaxCommands = 32
|
MaxCommands = 32
|
||||||
MaxDiagnosticBytes = 64 * 1024 // Lifetime budget; diagnostics are discarded.
|
MaxDiagnosticBytes = 64 * 1024 // Lifetime stderr budget; only structured diagnostics are logged.
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
|||||||
@@ -98,12 +98,14 @@ func (session *Session) authorize_remote_locked(role string, grants []string, pa
|
|||||||
access := &remoteAccess{action: hex.EncodeToString(token[:]), epoch: session.remoteEpoch,
|
access := &remoteAccess{action: hex.EncodeToString(token[:]), epoch: session.remoteEpoch,
|
||||||
role: role, grants: slices.Clone(grants), ctx: ctx, cancel: cancel}
|
role: role, grants: slices.Clone(grants), ctx: ctx, cancel: cancel}
|
||||||
session.remote = access
|
session.remote = access
|
||||||
|
session.log("tentacle remote authorized", "epoch", access.epoch, "role", access.role, "grants", access.grants)
|
||||||
// Parent cancellation invalidates delivery immediately. Retirement/event
|
// Parent cancellation invalidates delivery immediately. Retirement/event
|
||||||
// cleanup acquires the session lock afterwards and cannot retire a newer epoch.
|
// cleanup acquires the session lock afterwards and cannot retire a newer epoch.
|
||||||
access.stopPolicy = context.AfterFunc(parent, func() {
|
access.stopPolicy = context.AfterFunc(parent, func() {
|
||||||
session.mutex.Lock()
|
session.mutex.Lock()
|
||||||
defer session.mutex.Unlock()
|
defer session.mutex.Unlock()
|
||||||
if session.remote == access {
|
if session.remote == access {
|
||||||
|
session.log("tentacle remote policy cancelled", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -176,6 +178,7 @@ func (session *Session) revoke_remote_locked(reason string, notify bool) {
|
|||||||
if access == nil {
|
if access == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
session.log("tentacle remote revoked", "epoch", access.epoch, "reason", reason, "lease_open", access.lease != "", "sequence", access.sequence, "deadline_remaining_ms", time.Until(access.deadline).Milliseconds(), "context_cancelled", access.ctx.Err() != nil, "speech_items", access.speechItems, "key_items", access.keyItems)
|
||||||
session.remote = nil
|
session.remote = nil
|
||||||
access.cancel()
|
access.cancel()
|
||||||
access.speechBatch = nil
|
access.speechBatch = nil
|
||||||
@@ -281,6 +284,7 @@ func (session *Session) accept_child_request(frame []byte) error {
|
|||||||
value, failure := session.remote_call_locked(message.Method, message.Params)
|
value, failure := session.remote_call_locked(message.Method, message.Params)
|
||||||
reply := map[string]any{"jsonrpc": "2.0", "id": message.ID}
|
reply := map[string]any{"jsonrpc": "2.0", "id": message.ID}
|
||||||
if failure != nil {
|
if failure != nil {
|
||||||
|
session.log("tentacle session operation rejected", "operation", diagnosticOperation(message.Method), "code", failure.Code, "epoch", session.remoteEpoch)
|
||||||
reply["error"] = failure
|
reply["error"] = failure
|
||||||
} else {
|
} else {
|
||||||
reply["result"] = value
|
reply["result"] = value
|
||||||
@@ -324,6 +328,7 @@ func (session *Session) remote_call_locked(method string, params json.RawMessage
|
|||||||
}
|
}
|
||||||
access.lease, access.action, access.grants = hex.EncodeToString(token[:]), "", slices.Clone(request.Grants)
|
access.lease, access.action, access.grants = hex.EncodeToString(token[:]), "", slices.Clone(request.Grants)
|
||||||
session.refresh_remote_locked(RemoteHeartbeatTimeout)
|
session.refresh_remote_locked(RemoteHeartbeatTimeout)
|
||||||
|
session.log("tentacle remote lease opened", "epoch", access.epoch, "role", access.role, "heartbeat_timeout_ms", RemoteHeartbeatTimeout.Milliseconds())
|
||||||
return tentaclewire.RemoteLease{Lease: access.lease, Epoch: access.epoch}, nil
|
return tentaclewire.RemoteLease{Lease: access.lease, Epoch: access.epoch}, nil
|
||||||
case "remote.close", "remote.heartbeat":
|
case "remote.close", "remote.heartbeat":
|
||||||
var request tentaclewire.RemoteOperation
|
var request tentaclewire.RemoteOperation
|
||||||
|
|||||||
@@ -51,6 +51,7 @@ func (session *Session) remote_braille_locked(params json.RawMessage) (any, *RPC
|
|||||||
case session.remoteBraille <- RemoteBrailleDelivery{Context: access.ctx, Lease: lease, Role: access.role, Sequence: request.Sequence, Braille: request.Braille}:
|
case session.remoteBraille <- RemoteBrailleDelivery{Context: access.ctx, Lease: lease, Role: access.role, Sequence: request.Sequence, Braille: request.Braille}:
|
||||||
return struct{}{}, nil
|
return struct{}{}, nil
|
||||||
default:
|
default:
|
||||||
|
session.log("tentacle remote queue full", "queue", "braille", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
return nil, &RPCError{Code: -32002, Message: "Remote Braille queue full"}
|
return nil, &RPCError{Code: -32002, Message: "Remote Braille queue full"}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ func (session *Session) remote_focus_locked(params json.RawMessage) (any, *RPCEr
|
|||||||
return struct{}{}, nil
|
return struct{}{}, nil
|
||||||
default:
|
default:
|
||||||
// A lost "local" request must never leave capture active.
|
// A lost "local" request must never leave capture active.
|
||||||
|
session.log("tentacle remote queue full", "queue", "focus", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
return nil, &RPCError{Code: -32002, Message: "Remote focus queue full"}
|
return nil, &RPCError{Code: -32002, Message: "Remote focus queue full"}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -57,6 +57,7 @@ func (session *Session) remote_key_locked(params json.RawMessage) (any, *RPCErro
|
|||||||
data, _ := json.Marshal(request)
|
data, _ := json.Marshal(request)
|
||||||
bytes := len(data)
|
bytes := len(data)
|
||||||
if access.keyItems == tentaclewire.RemoteKeyQueueSize || access.keyBytes+bytes > tentaclewire.RemoteKeyQueueBytes {
|
if access.keyItems == tentaclewire.RemoteKeyQueueSize || access.keyBytes+bytes > tentaclewire.RemoteKeyQueueBytes {
|
||||||
|
session.log("tentacle remote queue full", "queue", "keyboard", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
return nil, &RPCError{Code: -32002, Message: "Remote keyboard queue full"}
|
return nil, &RPCError{Code: -32002, Message: "Remote keyboard queue full"}
|
||||||
}
|
}
|
||||||
@@ -79,6 +80,7 @@ func (session *Session) remote_key_locked(params json.RawMessage) (any, *RPCErro
|
|||||||
return struct{}{}, nil
|
return struct{}{}, nil
|
||||||
default:
|
default:
|
||||||
// Losing a release must never leave the input lease active.
|
// Losing a release must never leave the input lease active.
|
||||||
|
session.log("tentacle remote queue full", "queue", "keyboard", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
return nil, &RPCError{Code: -32002, Message: "Remote keyboard queue full"}
|
return nil, &RPCError{Code: -32002, Message: "Remote keyboard queue full"}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -115,6 +115,7 @@ func (session *Session) remote_present_locked(params json.RawMessage) (any, *RPC
|
|||||||
}
|
}
|
||||||
data, _ := json.Marshal(request.Speech)
|
data, _ := json.Marshal(request.Speech)
|
||||||
if access.speechItems == tentaclewire.RemoteSpeechQueueSize || access.speechBytes+len(data) > tentaclewire.RemoteSpeechQueueBytes {
|
if access.speechItems == tentaclewire.RemoteSpeechQueueSize || access.speechBytes+len(data) > tentaclewire.RemoteSpeechQueueBytes {
|
||||||
|
session.log("tentacle remote queue full", "queue", "speech", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
return nil, &RPCError{Code: -32002, Message: "Remote speech queue full"}
|
return nil, &RPCError{Code: -32002, Message: "Remote speech queue full"}
|
||||||
}
|
}
|
||||||
@@ -144,6 +145,7 @@ func (session *Session) remote_present_locked(params json.RawMessage) (any, *RPC
|
|||||||
return struct{}{}, nil
|
return struct{}{}, nil
|
||||||
default:
|
default:
|
||||||
// A dropped cancel would let stale speech outlive the intended stream.
|
// A dropped cancel would let stale speech outlive the intended stream.
|
||||||
|
session.log("tentacle remote queue full", "queue", "speech", "epoch", access.epoch)
|
||||||
session.revoke_remote_locked("local", true)
|
session.revoke_remote_locked("local", true)
|
||||||
return nil, &RPCError{Code: -32002, Message: "Remote speech queue full"}
|
return nil, &RPCError{Code: -32002, Message: "Remote speech queue full"}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
|
"log/slog"
|
||||||
"strconv"
|
"strconv"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
@@ -48,6 +49,7 @@ type outgoing struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type Session struct {
|
type Session struct {
|
||||||
|
logger *slog.Logger
|
||||||
outputContext context.Context
|
outputContext context.Context
|
||||||
revokeOutput context.CancelFunc
|
revokeOutput context.CancelFunc
|
||||||
|
|
||||||
@@ -136,6 +138,7 @@ func (session *Session) Activate() error {
|
|||||||
}
|
}
|
||||||
session.bindings = resolve_bindings(session.spec.ID, session.commands, occupied)
|
session.bindings = resolve_bindings(session.spec.ID, session.commands, occupied)
|
||||||
session.state = Active
|
session.state = Active
|
||||||
|
session.log("tentacle activated")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -251,6 +254,9 @@ func (session *Session) finish_locked(id string, value json.RawMessage, err erro
|
|||||||
err = contextError
|
err = contextError
|
||||||
value = nil
|
value = nil
|
||||||
}
|
}
|
||||||
|
if err != nil {
|
||||||
|
session.log("tentacle request failed", "request", id, "reason", diagnosticError(err))
|
||||||
|
}
|
||||||
call.cancel()
|
call.cancel()
|
||||||
<-session.manager.pending
|
<-session.manager.pending
|
||||||
call.result <- Result{Session: session.identity, ID: id, Value: value, Err: err}
|
call.result <- Result{Session: session.identity, ID: id, Value: value, Err: err}
|
||||||
@@ -294,6 +300,7 @@ func (session *Session) fail_locked(err error) {
|
|||||||
if session.state == Stopping || session.state == Stopped || session.state == Failed {
|
if session.state == Stopping || session.state == Stopped || session.state == Failed {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
session.log("tentacle failed", "state", session.state, "reason", diagnosticError(err), "pending", len(session.pending), "queued_bytes", session.queuedBytes)
|
||||||
session.state, session.failure = Failed, err
|
session.state, session.failure = Failed, err
|
||||||
close(session.revoked)
|
close(session.revoked)
|
||||||
session.revokeOutput()
|
session.revokeOutput()
|
||||||
@@ -321,6 +328,7 @@ func (session *Session) Disable() {
|
|||||||
if session.state == Stopping || session.state == Stopped || session.state == Failed {
|
if session.state == Stopping || session.state == Stopped || session.state == Failed {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
session.log("tentacle disabling", "state", session.state, "pending", len(session.pending))
|
||||||
session.state = Stopping
|
session.state = Stopping
|
||||||
close(session.revoked)
|
close(session.revoked)
|
||||||
session.revokeOutput()
|
session.revokeOutput()
|
||||||
|
|||||||
Reference in new issue
Block a user