From 684981c235128406c252d079b4f93ae80669c520 Mon Sep 17 00:00:00 2001 From: cabbage Date: Fri, 11 Sep 2026 08:56:00 +0000 Subject: [PATCH] fix: order command controls after dispatch --- cmd/rvbox-server/main.go | 15 ++- cmd/rvbox/main.go | 11 ++ deploy/production/server.toml.example | 6 + internal/client/agent/runner.go | 36 ++--- internal/config/defaults.go | 2 +- internal/config/effective_test.go | 2 +- internal/observability/rotate.go | 156 ++++++++++++++++++++++ internal/observability/rotate_test.go | 86 ++++++++++++ internal/server/session/agent_server.go | 48 +++++-- internal/server/session/session.go | 4 + internal/server/store/stdin.go | 12 +- test/coverage.toml | 18 +++ test/windowsnative/native_fixture_test.go | 2 +- 13 files changed, 360 insertions(+), 38 deletions(-) create mode 100644 internal/observability/rotate.go create mode 100644 internal/observability/rotate_test.go diff --git a/cmd/rvbox-server/main.go b/cmd/rvbox-server/main.go index a4257a9..6bd02b4 100644 --- a/cmd/rvbox-server/main.go +++ b/cmd/rvbox-server/main.go @@ -6,6 +6,7 @@ import ( "errors" "flag" "fmt" + "io" "log" "net" "net/http" @@ -58,6 +59,12 @@ func run(configPath string) error { if err != nil { return err } + runtimeLog, err := observability.OpenRotatingFile(configured.Observability.LogFile, configured.Observability.LogMaxBytes, configured.Observability.LogMaxFiles) + if err != nil { + return fmt.Errorf("open server rotating log: %w", err) + } + defer runtimeLog.Close() + logger := log.New(io.MultiWriter(os.Stderr, observability.FormatLog(runtimeLog, configured.Observability.LogFormat)), "rvbox-server: ", log.LstdFlags|log.LUTC) persistence, err := store.Open(context.Background(), store.Options{ DataDir: configured.Server.DataDir, BusyTimeout: configured.Storage.SQLiteBusyTimeout, SegmentTargetSize: configured.Storage.SegmentTargetBytes, @@ -84,7 +91,7 @@ func run(configPath string) error { if configured.Observability.Listen != "" { healthListener, err = net.Listen("tcp", configured.Observability.Listen) if err != nil { - log.Printf("rvbox-server: observability endpoint unavailable (continuing without it): %v", err) + logger.Printf("observability endpoint unavailable (continuing without it): %v", err) } } if healthListener != nil { @@ -104,10 +111,10 @@ func run(configPath string) error { Scope: store.IncidentScopeGlobal, ScopeKey: "server-startup-recovery", Summary: "startup storage recovery failed", Evidence: []byte(recoverErr.Error()), AutomaticallyRepairable: false, }); incidentErr != nil { - log.Printf("rvbox-server: could not persist recovery incident: %v", incidentErr) + logger.Printf("could not persist recovery incident: %v", incidentErr) } } - log.Printf("rvbox-server: storage recovery left readiness disabled: %v", recoverErr) + logger.Printf("storage recovery left readiness disabled: %v", recoverErr) return } health.SetReady(true) @@ -143,7 +150,7 @@ func run(configPath string) error { } defer rpcListener.Close() if configured.JSONRPC.NonLoopbackBind { - log.Printf("rvbox-server: WARNING JSON-RPC is unauthenticated and bound to non-loopback address %s", configured.JSONRPC.Listen) + logger.Printf("WARNING JSON-RPC is unauthenticated and bound to non-loopback address %s", configured.JSONRPC.Listen) } rpcServer = &http.Server{Handler: control.NewJSONRPCHandler(controlService, int64(configured.Protocol.MaxJSONRPCBodyBytes)), ReadHeaderTimeout: configured.Flow.WriteDeadline} } diff --git a/cmd/rvbox/main.go b/cmd/rvbox/main.go index f60176e..e9fe1b5 100644 --- a/cmd/rvbox/main.go +++ b/cmd/rvbox/main.go @@ -150,6 +150,17 @@ func runClientDaemon(ctx context.Context, configPath string, diagnostics io.Writ if err != nil { return err } + runtimeLog, err := observability.OpenRotatingFile(configured.Observability.LogFile, configured.Observability.LogMaxBytes, configured.Observability.LogMaxFiles) + if err != nil { + return fmt.Errorf("open client rotating log: %w", err) + } + defer runtimeLog.Close() + formattedRuntimeLog := observability.FormatLog(runtimeLog, configured.Observability.LogFormat) + if diagnostics == nil { + diagnostics = formattedRuntimeLog + } else { + diagnostics = io.MultiWriter(formattedRuntimeLog, diagnostics) + } health := observability.New() go func() { if serveErr := health.Serve(ctx, configured.Observability.Listen, observability.Paths{Liveness: configured.Observability.LivenessPath, Readiness: configured.Observability.ReadinessPath, Metrics: configured.Observability.MetricsPath}); serveErr != nil && ctx.Err() == nil && diagnostics != nil { diff --git a/deploy/production/server.toml.example b/deploy/production/server.toml.example index c694eca..981ec24 100644 --- a/deploy/production/server.toml.example +++ b/deploy/production/server.toml.example @@ -19,3 +19,9 @@ enabled = false [observability] # nginx exposes only /livez and /readyz, not the metrics listener itself. listen = "0.0.0.0:6901" +# The server retains its own bounded diagnostics beside durable state. +log_file = "/var/lib/rvbox-server/logs/rvbox-server.log" +# Rotate the current log after this many bytes (10 MiB). +log_max_bytes = 10485760 +# Number of sealed rotated log files to retain. +log_max_files = 5 diff --git a/internal/client/agent/runner.go b/internal/client/agent/runner.go index f56054e..1b5ac69 100644 --- a/internal/client/agent/runner.go +++ b/internal/client/agent/runner.go @@ -159,7 +159,7 @@ func RunOnce(ctx context.Context, options RunnerOptions) error { func runOnce(ctx context.Context, options RunnerOptions) (resultErr error) { transport, err := options.Dial(ctx) if err != nil { - return err + return fmt.Errorf("dial agent server: %w", err) } defer func() { if closeErr := transport.Close(); resultErr == nil && closeErr != nil { @@ -173,7 +173,7 @@ func runOnce(ctx context.Context, options RunnerOptions) (resultErr error) { } session, err := Handshake(ctx, transport, options.Hello, limits) if err != nil { - return err + return fmt.Errorf("agent handshake: %w", err) } snapshot, err := options.Store.ReconcileSnapshot(ctx) if err != nil { @@ -181,7 +181,7 @@ func runOnce(ctx context.Context, options RunnerOptions) (resultErr error) { } result, err := Reconcile(ctx, transport, session, snapshot, limits) if err != nil { - return err + return fmt.Errorf("exchange reconciliation: %w", err) } terminated, err := ApplyReconcileResult(ctx, options.Store, result, options.Now()) if err != nil { @@ -190,16 +190,16 @@ func runOnce(ctx context.Context, options RunnerOptions) (resultErr error) { for _, issue := range terminated { if options.OnTerminate != nil { if err := options.OnTerminate(ctx, issue); err != nil { - return err + return fmt.Errorf("terminate reconciled command %s: %w", issue, err) } } } sentEvents := make(map[domain.UUID]uint64) if err := replayEvents(ctx, transport, options.Store, session, snapshot, limits, sentEvents); err != nil { - return err + return fmt.Errorf("replay client events: %w", err) } if err := sendCapacity(ctx, transport, session, options.Hello, snapshot, limits); err != nil { - return err + return fmt.Errorf("advertise client capacity: %w", err) } return serveActive(ctx, transport, options, session, sentEvents) } @@ -330,20 +330,20 @@ func serveActive(ctx context.Context, transport Transport, options RunnerOptions case <-ctx.Done(): return nil case err := <-readErrors: - return err + return fmt.Errorf("read active agent frame: %w", err) case issue := <-options.EventReady: if issue == (domain.UUID{}) { continue } if err := flushEvents(ctx, transport, options.Store, session, issue, limits, sent); err != nil { - return err + return fmt.Errorf("flush ready command %s events: %w", issue, err) } continue case encoded = <-frames: } envelope, err := agentproto.DecodeEnvelope(encoded, limits, rvboxv1.Platform_PLATFORM_WINDOWS) if err != nil { - return err + return fmt.Errorf("decode active agent frame: %w", err) } if envelope.GetSessionId() != session.ID || envelope.GetSessionGeneration() != session.Generation { return ErrProtocolHandshake @@ -365,11 +365,11 @@ func serveActive(ctx context.Context, transport Transport, options RunnerOptions } case envelope.GetScriptChunk() != nil: if err := handleScriptChunk(ctx, transport, options.Store, session, envelope.GetScriptChunk(), limits, options.Now, sent); err != nil { - return err + return fmt.Errorf("apply script chunk for %s: %w", envelope.GetScriptChunk().GetIssueUuid(), err) } case envelope.GetScriptCommit() != nil: if err := handleScriptCommit(ctx, transport, options.Store, session, envelope.GetScriptCommit(), limits, options.Now, sent); err != nil { - return err + return fmt.Errorf("commit script for %s: %w", envelope.GetScriptCommit().GetIssueUuid(), err) } if options.OnScriptReady != nil { issue, err := domain.ParseUUIDv7(envelope.GetScriptCommit().GetIssueUuid()) @@ -377,40 +377,40 @@ func serveActive(ctx context.Context, transport Transport, options RunnerOptions return err } if err := options.OnScriptReady(ctx, session, issue); err != nil { - return err + return fmt.Errorf("start committed script %s: %w", issue, err) } } case envelope.GetStdinWrite() != nil: if options.OnStdin != nil { if err := options.OnStdin(ctx, session, envelope.GetStdinWrite()); err != nil { - return err + return fmt.Errorf("append stdin for %s: %w", envelope.GetStdinWrite().GetIssueUuid(), err) } } if issue, parseErr := domain.ParseUUIDv7(envelope.GetStdinWrite().GetIssueUuid()); parseErr == nil { if err := flushEvents(ctx, transport, options.Store, session, issue, limits, sent); err != nil { - return err + return fmt.Errorf("flush stdin command %s events: %w", issue, err) } } case envelope.GetCloseStdin() != nil: if options.OnCloseStdin != nil { if err := options.OnCloseStdin(ctx, session, envelope.GetCloseStdin()); err != nil { - return err + return fmt.Errorf("close stdin for %s: %w", envelope.GetCloseStdin().GetIssueUuid(), err) } } if issue, parseErr := domain.ParseUUIDv7(envelope.GetCloseStdin().GetIssueUuid()); parseErr == nil { if err := flushEvents(ctx, transport, options.Store, session, issue, limits, sent); err != nil { - return err + return fmt.Errorf("flush closed-stdin command %s events: %w", issue, err) } } case envelope.GetSignalCommand() != nil: if options.OnSignal != nil { if err := options.OnSignal(ctx, session, envelope.GetSignalCommand()); err != nil { - return err + return fmt.Errorf("signal command %s: %w", envelope.GetSignalCommand().GetIssueUuid(), err) } } if issue, parseErr := domain.ParseUUIDv7(envelope.GetSignalCommand().GetIssueUuid()); parseErr == nil { if err := flushEvents(ctx, transport, options.Store, session, issue, limits, sent); err != nil { - return err + return fmt.Errorf("flush signalled command %s events: %w", issue, err) } } case envelope.GetError() != nil: diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 34aedbc..f12f3f7 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -40,7 +40,7 @@ func defaultServerFile() serverFile { }, Observability: observabilityFile{ Listen: "127.0.0.1:6901", LivenessPath: "/livez", ReadinessPath: "/readyz", - MetricsPath: "/metrics", LogLevel: "info", LogFormat: "json", + MetricsPath: "/metrics", LogLevel: "info", LogFormat: "json", LogMaxBytes: 10 << 20, LogMaxFiles: 5, }, } } diff --git a/internal/config/effective_test.go b/internal/config/effective_test.go index b3a78db..049ff80 100644 --- a/internal/config/effective_test.go +++ b/internal/config/effective_test.go @@ -23,7 +23,7 @@ func TestEffectiveConfigurationGolden_HP_CFG_09(t *testing.T) { t.Fatal("effective server configuration is nondeterministic") } digest := sha256.Sum256(first) - const goldenSHA256 = "55c5fec2c10bc27372353a60f317c5a2959f8087685cf7c11202bb12dc59aa28" + const goldenSHA256 = "65c938efd935ce0c0d2354cdb62c38a2dc177755520d1ae1e5951902a70dd323" if got := hex.EncodeToString(digest[:]); got != goldenSHA256 { t.Fatalf("effective server configuration golden changed: got %s", got) } diff --git a/internal/observability/rotate.go b/internal/observability/rotate.go new file mode 100644 index 0000000..492f516 --- /dev/null +++ b/internal/observability/rotate.go @@ -0,0 +1,156 @@ +package observability + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "sync" + "time" +) + +// RotatingFile is an append-only daemon log with bounded numbered archives. +// It is deliberately a single-process writer: deployment supplies one daemon +// process per configured path, and a second writer must use a distinct file. +// Archive .1 is newest; .maxFiles is oldest. +type RotatingFile struct { + mu sync.Mutex + path string + maxBytes uint64 + maxFiles uint32 + file *os.File + size uint64 +} + +// OpenRotatingFile opens path for append, creating its parent directories with +// conservative permissions. A blank path disables file logging and returns a +// no-op closer. Both rotation controls must be zero (unbounded) or positive. +func OpenRotatingFile(path string, maxBytes uint64, maxFiles uint32) (io.WriteCloser, error) { + if path == "" { + return nopWriteCloser{Writer: io.Discard}, nil + } + if (maxBytes == 0) != (maxFiles == 0) { + return nil, errors.New("log rotation byte/file limits must both be zero or positive") + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return nil, fmt.Errorf("create log directory: %w", err) + } + result := &RotatingFile{path: path, maxBytes: maxBytes, maxFiles: maxFiles} + if err := result.open(); err != nil { + return nil, err + } + return result, nil +} + +func (file *RotatingFile) open() error { + opened, err := os.OpenFile(file.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600) + if err != nil { + return fmt.Errorf("open log file: %w", err) + } + info, err := opened.Stat() + if err != nil { + _ = opened.Close() + return fmt.Errorf("stat log file: %w", err) + } + file.file = opened + file.size = uint64(info.Size()) + return nil +} + +func (file *RotatingFile) Write(data []byte) (int, error) { + file.mu.Lock() + defer file.mu.Unlock() + if file.file == nil { + return 0, os.ErrClosed + } + if file.maxBytes != 0 && file.size != 0 && (file.size >= file.maxBytes || uint64(len(data)) > file.maxBytes-file.size) { + if err := file.rotate(); err != nil { + return 0, err + } + } + written, err := file.file.Write(data) + file.size += uint64(written) + return written, err +} + +func (file *RotatingFile) Close() error { + file.mu.Lock() + defer file.mu.Unlock() + if file.file == nil { + return nil + } + err := file.file.Close() + file.file = nil + return err +} + +func (file *RotatingFile) rotate() error { + if err := file.file.Close(); err != nil { + return fmt.Errorf("close log before rotation: %w", err) + } + file.file = nil + oldest := file.archivePath(file.maxFiles) + if err := os.Remove(oldest); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("remove oldest log archive: %w", err) + } + for index := file.maxFiles; index > 1; index-- { + from := file.archivePath(index - 1) + to := file.archivePath(index) + if err := os.Rename(from, to); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("rotate log archive: %w", err) + } + } + if err := os.Rename(file.path, file.archivePath(1)); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("seal current log: %w", err) + } + return file.open() +} + +func (file *RotatingFile) archivePath(index uint32) string { + return file.path + "." + strconv.FormatUint(uint64(index), 10) +} + +type nopWriteCloser struct{ io.Writer } + +func (nopWriteCloser) Close() error { return nil } + +// FormatLog makes file records either newline-delimited JSON or plain text. +// It is intentionally applied only to daemon diagnostics, never command +// output, stdin, environment values, or other payload-bearing data. +func FormatLog(destination io.Writer, format string) io.Writer { + if destination == nil || format != "json" { + return destination + } + return &jsonLogWriter{destination: destination} +} + +type jsonLogWriter struct { + mu sync.Mutex + destination io.Writer +} + +func (writer *jsonLogWriter) Write(data []byte) (int, error) { + writer.mu.Lock() + defer writer.mu.Unlock() + for _, line := range bytes.Split(data, []byte{'\n'}) { + if len(line) == 0 { + continue + } + record, err := json.Marshal(struct { + Time string `json:"time"` + Level string `json:"level"` + Message string `json:"message"` + }{Time: time.Now().UTC().Format(time.RFC3339Nano), Level: "info", Message: string(line)}) + if err != nil { + return 0, fmt.Errorf("encode structured log: %w", err) + } + if _, err := writer.destination.Write(append(record, '\n')); err != nil { + return 0, err + } + } + return len(data), nil +} diff --git a/internal/observability/rotate_test.go b/internal/observability/rotate_test.go new file mode 100644 index 0000000..b443107 --- /dev/null +++ b/internal/observability/rotate_test.go @@ -0,0 +1,86 @@ +package observability + +import ( + "bytes" + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestRotatingFileBoundsArchives_HP_OPS_02(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "nested", "rvbox.log") + writer, err := OpenRotatingFile(path, 5, 2) + if err != nil { + t.Fatal(err) + } + if _, err := writer.Write([]byte("first")); err != nil { + t.Fatal(err) + } + if _, err := writer.Write([]byte("two")); err != nil { + t.Fatal(err) + } + if _, err := writer.Write([]byte("three")); err != nil { + t.Fatal(err) + } + if _, err := writer.Write([]byte("four")); err != nil { + t.Fatal(err) + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + current, err := os.ReadFile(path) + if err != nil || string(current) != "four" { + t.Fatalf("current = %q, %v", current, err) + } + newest, err := os.ReadFile(path + ".1") + if err != nil || string(newest) != "three" { + t.Fatalf("newest archive = %q, %v", newest, err) + } + oldest, err := os.ReadFile(path + ".2") + if err != nil || string(oldest) != "two" { + t.Fatalf("oldest archive = %q, %v", oldest, err) + } + if _, err := os.Stat(path + ".3"); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("unexpected third archive: %v", err) + } +} + +func TestRotatingFileRejectsPartialRotationConfig_BH_OPS_03(t *testing.T) { + t.Parallel() + for _, limits := range [][2]uint64{{1, 0}, {0, 1}} { + _, err := OpenRotatingFile(filepath.Join(t.TempDir(), "rvbox.log"), limits[0], uint32(limits[1])) + if err == nil || !strings.Contains(err.Error(), "both be zero") { + t.Fatalf("limits %v error = %v", limits, err) + } + } +} + +func TestFormatLogWritesStructuredBoundedRecords_HP_OPS_04(t *testing.T) { + t.Parallel() + var structured bytes.Buffer + if _, err := FormatLog(&structured, "json").Write([]byte("connected\nrecovered\n")); err != nil { + t.Fatal(err) + } + lines := bytes.Split(bytes.TrimSpace(structured.Bytes()), []byte{'\n'}) + if len(lines) != 2 { + t.Fatalf("structured records = %q", structured.String()) + } + for index, want := range []string{"connected", "recovered"} { + var record struct { + Time string `json:"time"` + Level string `json:"level"` + Message string `json:"message"` + } + if err := json.Unmarshal(lines[index], &record); err != nil || record.Time == "" || record.Level != "info" || record.Message != want { + t.Fatalf("record %d = %+v, %v", index, record, err) + } + } + var plain bytes.Buffer + if _, err := FormatLog(&plain, "text").Write([]byte("plain\n")); err != nil || plain.String() != "plain\n" { + t.Fatalf("text log = %q, %v", plain.String(), err) + } +} diff --git a/internal/server/session/agent_server.go b/internal/server/session/agent_server.go index 3c8277f..fa28ee7 100644 --- a/internal/server/session/agent_server.go +++ b/internal/server/session/agent_server.go @@ -180,7 +180,7 @@ func (server *AgentServer) serveConnection(parent context.Context, connection *w server.close(connection, websocket.StatusPolicyViolation, "invalid initial client capacity") return } - go server.dispatchLoop(sessionContext, queue, handle.DispatchWake(), reconciled, &capacityMu, capacity, reservations, stdinSent, signalSent, scriptTransfers, hello.GetClientId(), hello.GetPlatform(), encodeSessionID(sessionID), registration.Generation, cancel) + go server.dispatchLoop(sessionContext, queue, handle.DispatchWake(), handle.SignalDispatch, reconciled, &capacityMu, capacity, reservations, stdinSent, signalSent, scriptTransfers, hello.GetClientId(), hello.GetPlatform(), encodeSessionID(sessionID), registration.Generation, cancel) encodedSessionID := encodeSessionID(sessionID) welcome, err := proto.Marshal(&rvboxv1.AgentEnvelope{ SessionId: encodedSessionID, SessionGeneration: registration.Generation, @@ -413,7 +413,7 @@ func eventType(event *rvboxv1.CommandEvent) uint16 { // enqueueNextDispatch records the queued-to-dispatched transition before // exposing work to the network. A full data lane is a pre-write failure, so // only the owning generation can put the command back into the queue. -func (server *AgentServer) enqueueNextDispatch(ctx context.Context, queue *WriterQueue, clientID string, platform rvboxv1.Platform, sessionID string, generation uint64, beforeEnqueue func(*store.DispatchCandidate)) (domain.UUID, bool, error) { +func (server *AgentServer) enqueueNextDispatch(ctx context.Context, queue *WriterQueue, clientID string, platform rvboxv1.Platform, sessionID string, generation uint64, beforeEnqueue func(*store.DispatchCandidate), onWritten func(domain.UUID)) (domain.UUID, bool, error) { candidate, err := server.Store.ClaimNextDispatch(ctx, clientID, generation, server.now()) if err != nil || candidate == nil { return domain.UUID{}, false, err @@ -451,7 +451,11 @@ func (server *AgentServer) enqueueNextDispatch(ctx context.Context, queue *Write if beforeEnqueue != nil { beforeEnqueue(candidate) } - if !queue.EnqueueData(Frame{Kind: FrameData, Payload: encoded}) { + if !queue.EnqueueData(Frame{Kind: FrameData, Payload: encoded, OnWritten: func() { + if onWritten != nil { + onWritten(candidate.IssueUUID) + } + }}) { sent, requeueErr := requeue(ErrDispatchDataFull) return candidate.IssueUUID, sent, requeueErr } @@ -531,12 +535,18 @@ func (server *AgentServer) enqueueNextScript(queue *WriterQueue, sessionID strin // lane. Session-local sent tracking suppresses duplicate frames while a live // connection remains usable; reconnecting naturally replays unacknowledged // writes from storage. -func (server *AgentServer) enqueueNextStdin(ctx context.Context, queue *WriterQueue, clientID, sessionID string, generation uint64, sent map[string]struct{}, sentMu *sync.Mutex) (bool, error) { +func (server *AgentServer) enqueueNextStdin(ctx context.Context, queue *WriterQueue, clientID, sessionID string, generation uint64, sent map[string]struct{}, dispatchWritten map[string]bool, sentMu *sync.Mutex) (bool, error) { intents, err := server.Store.PendingStdin(ctx, clientID) if err != nil { return false, err } for _, intent := range intents { + sentMu.Lock() + written, waitingForDispatch := dispatchWritten[intent.IssueUUID.String()] + sentMu.Unlock() + if waitingForDispatch && !written { + continue + } key := stdinIntentKey(intent.IssueUUID, intent.WriteSeq) sentMu.Lock() _, alreadySent := sent[key] @@ -579,12 +589,18 @@ func signalIntentKey(issue domain.UUID, revision uint64, signal rvboxv1.SignalKi return issue.String() + ":" + fmt.Sprint(revision) + ":" + fmt.Sprint(signal) } -func (server *AgentServer) enqueueNextSignal(ctx context.Context, queue *WriterQueue, clientID, sessionID string, generation uint64, sent map[string]struct{}, sentMu *sync.Mutex) (bool, error) { +func (server *AgentServer) enqueueNextSignal(ctx context.Context, queue *WriterQueue, clientID, sessionID string, generation uint64, sent map[string]struct{}, dispatchWritten map[string]bool, sentMu *sync.Mutex) (bool, error) { intents, err := server.Store.PendingSignals(ctx, clientID, generation) if err != nil { return false, err } for _, intent := range intents { + sentMu.Lock() + written, waitingForDispatch := dispatchWritten[intent.IssueUUID.String()] + sentMu.Unlock() + if waitingForDispatch && !written { + continue + } key := signalIntentKey(intent.IssueUUID, intent.CommandRevision, intent.Signal) sentMu.Lock() _, alreadySent := sent[key] @@ -617,7 +633,11 @@ func (server *AgentServer) enqueueNextSignal(ctx context.Context, queue *WriterQ // dispatchLoop is the per-session serialized dispatcher. It waits for a // complete reconciliation result before consuming queued work, then coalesces // wakeups from local control RPCs, capacity advertisements, and acceptances. -func (server *AgentServer) dispatchLoop(ctx context.Context, queue *WriterQueue, wake <-chan struct{}, reconciled <-chan struct{}, capacityMu *sync.Mutex, capacity *CapacityShadow, reservations map[string]DispatchLane, stdinSent, signalSent map[string]struct{}, scriptTransfers map[string]*scriptTransfer, clientID string, platform rvboxv1.Platform, sessionID string, generation uint64, cancel context.CancelFunc) { +func (server *AgentServer) dispatchLoop(ctx context.Context, queue *WriterQueue, wake <-chan struct{}, signalWake func(), reconciled <-chan struct{}, capacityMu *sync.Mutex, capacity *CapacityShadow, reservations map[string]DispatchLane, stdinSent, signalSent map[string]struct{}, scriptTransfers map[string]*scriptTransfer, clientID string, platform rvboxv1.Platform, sessionID string, generation uint64, cancel context.CancelFunc) { + // A control frame may use the essential queue and therefore overtake data. + // Keep its issue blocked only until the dispatch frame has actually crossed + // the socket writer; after that, WebSocket ordering preserves the dependency. + dispatchWritten := make(map[string]bool) select { case <-reconciled: case <-ctx.Done(): @@ -625,7 +645,7 @@ func (server *AgentServer) dispatchLoop(ctx context.Context, queue *WriterQueue, } for { for { - stdinQueued, stdinErr := server.enqueueNextStdin(ctx, queue, clientID, sessionID, generation, stdinSent, capacityMu) + stdinQueued, stdinErr := server.enqueueNextStdin(ctx, queue, clientID, sessionID, generation, stdinSent, dispatchWritten, capacityMu) if stdinErr != nil { server.closeForDispatchFailure(cancel) return @@ -633,7 +653,7 @@ func (server *AgentServer) dispatchLoop(ctx context.Context, queue *WriterQueue, if stdinQueued { continue } - signalQueued, signalErr := server.enqueueNextSignal(ctx, queue, clientID, sessionID, generation, signalSent, capacityMu) + signalQueued, signalErr := server.enqueueNextSignal(ctx, queue, clientID, sessionID, generation, signalSent, dispatchWritten, capacityMu) if signalErr != nil { server.closeForDispatchFailure(cancel) return @@ -659,16 +679,25 @@ func (server *AgentServer) dispatchLoop(ctx context.Context, queue *WriterQueue, issue, sent, err := server.enqueueNextDispatch(ctx, queue, clientID, platform, sessionID, generation, func(candidate *store.DispatchCandidate) { capacityMu.Lock() reservations[candidate.IssueUUID.String()] = lane + dispatchWritten[candidate.IssueUUID.String()] = false if candidate.ScriptPresent { scriptTransfers[candidate.IssueUUID.String()] = &scriptTransfer{Body: append([]byte(nil), candidate.ScriptContent...), Digest: sha256.Sum256(candidate.ScriptContent)} } inserted = true capacityMu.Unlock() + }, func(issue domain.UUID) { + capacityMu.Lock() + dispatchWritten[issue.String()] = true + capacityMu.Unlock() + if signalWake != nil { + signalWake() + } }) if !sent { capacityMu.Lock() if inserted { delete(reservations, issue.String()) + delete(dispatchWritten, issue.String()) delete(scriptTransfers, issue.String()) } capacity.Release(lane) @@ -710,6 +739,9 @@ func (server *AgentServer) writeLoop(ctx context.Context, connection *websocket. if frame.Written != nil { close(frame.Written) } + if frame.OnWritten != nil { + frame.OnWritten() + } continue } if !errors.Is(err, context.DeadlineExceeded) { diff --git a/internal/server/session/session.go b/internal/server/session/session.go index 5b35acb..60b6d0c 100644 --- a/internal/server/session/session.go +++ b/internal/server/session/session.go @@ -26,6 +26,10 @@ type Frame struct { // Written is closed by the sole socket writer after a successful write. // It is used only for protocol barriers such as reconciliation-before-work. Written chan<- struct{} + // OnWritten is a non-blocking session-local scheduler notification. It is + // invoked only after the sole socket writer has completed the frame, so a + // dependent control frame cannot overtake its command dispatch. + OnWritten func() } // WriterQueue is owned by one socket writer. Data saturation leaves the work diff --git a/internal/server/store/stdin.go b/internal/server/store/stdin.go index 3d24d30..56156ab 100644 --- a/internal/server/store/stdin.go +++ b/internal/server/store/stdin.go @@ -46,10 +46,12 @@ type StdinIntent struct { Close bool } -// PendingStdin returns unacknowledged input intents for one client. Delivery is -// intentionally tracked by the session dispatcher, not by this durable query: -// a lost connection simply causes the next session to replay the same write -// sequence, which the client acknowledges idempotently. +// PendingStdin returns unacknowledged input intents only after the command has +// entered DISPATCHED state. This durable gate prevents append/close controls +// from overtaking their initial CommandDispatch on a fresh session. Delivery is +// otherwise tracked by the session dispatcher: a lost connection simply causes +// the next session to replay the same write sequence, which the client +// acknowledges idempotently. func (store *Store) PendingStdin(ctx context.Context, clientID string) ([]StdinIntent, error) { if clientID == "" { return nil, errors.New("client ID is required") @@ -60,7 +62,7 @@ func (store *Store) PendingStdin(ctx context.Context, clientID string) ([]StdinI } rows, err := database.QueryContext(ctx, `SELECT w.issue_uuid, w.write_seq, w.payload, w.raw_bytes, w.stored_bytes, w.compression, w.sha256, w.append_newline, w.close_intent FROM stdin_writes w JOIN commands c ON c.issue_uuid = w.issue_uuid -WHERE c.client_id = ? AND c.lifecycle BETWEEN 1 AND 4 AND w.acknowledged = 0 +WHERE c.client_id = ? AND c.lifecycle BETWEEN 2 AND 4 AND w.acknowledged = 0 ORDER BY c.issue_time, c.issue_uuid, w.write_seq`, clientID) if err != nil { return nil, err diff --git a/test/coverage.toml b/test/coverage.toml index a4f819f..d470901 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -619,6 +619,24 @@ layer = "unit" status = "implemented" tests = ["test/windowsnative/native_fixture_test.go:TestProductionComposeAssets_HP_OPS_01"] +[[requirements]] +id = "HP-OPS-02" +layer = "unit" +status = "implemented" +tests = ["internal/observability/rotate_test.go:TestRotatingFileBoundsArchives_HP_OPS_02"] + +[[requirements]] +id = "BH-OPS-03" +layer = "unit" +status = "implemented" +tests = ["internal/observability/rotate_test.go:TestRotatingFileRejectsPartialRotationConfig_BH_OPS_03"] + +[[requirements]] +id = "HP-OPS-04" +layer = "unit" +status = "implemented" +tests = ["internal/observability/rotate_test.go:TestFormatLogWritesStructuredBoundedRecords_HP_OPS_04"] + [[requirements]] id = "HP-STORE-01" layer = "integration" diff --git a/test/windowsnative/native_fixture_test.go b/test/windowsnative/native_fixture_test.go index 881b6c9..f73072e 100644 --- a/test/windowsnative/native_fixture_test.go +++ b/test/windowsnative/native_fixture_test.go @@ -102,7 +102,7 @@ func TestProductionComposeAssets_HP_OPS_01(t *testing.T) { } } config := read("deploy/production/server.toml.example") - for _, required := range []string{"agent_listen = \"0.0.0.0:6899\"", "listen = \"0.0.0.0:6901\"", "enabled = false"} { + for _, required := range []string{"agent_listen = \"0.0.0.0:6899\"", "listen = \"0.0.0.0:6901\"", "enabled = false", "log_file = \"/var/lib/rvbox-server/logs/rvbox-server.log\"", "log_max_bytes = 10485760", "log_max_files = 5"} { if !strings.Contains(config, required) { t.Fatalf("production server example is missing %q", required) }