mirror of
https://github.com/NVIDIA/OpenShell.git
synced 2026-10-02 07:34:45 +08:00
fix(api): make WatchSandbox loss-aware and resumable (#3209)
* fix(api): emit warning on WatchSandbox broadcast lag instead of terminating Broadcast lag on the status, log, and platform receivers was converted to a RESOURCE_EXHAUSTED status that terminated the whole watch stream. Lag is recoverable: the receiver resumes at the oldest surviving message. Emit a SandboxStreamWarning and continue streaming instead; keep terminating on Closed. Add helpers and unit tests covering the warning payload and receiver recovery after lag. Partially addresses #3055 (cursor/resume follow up separately). Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * refactor(server): group per-sandbox log bus state and stamp sequence numbers Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * feat(proto): add resume cursor fields to sandbox watch API Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * feat(server): stamp watch cursors from a shared per-sandbox sequence Allocate cursors from a single SeqAllocator shared by the log and platform event buses, so a sandbox's merged watch stream carries unique, strictly increasing cursors. A single resume_after_cursor can then unambiguously locate a client's position across both sources. Rewrite both publish paths to allocate the sequence, stamp event.cursor, send, and append to the tail under one lock. This removes the previous get_mut().expect() TOCTOU race where a concurrent remove() between the two lock sections could panic. Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * feat(server): serve WatchSandbox resume from cursor with gap detection Add tail_after() to the log and platform event buses, returning every buffered event newer than a client's resume cursor. Each PerSandbox now tracks last_trimmed_seq (the highest seq it has evicted) so a resume is reported as an unrecoverable ResumeGap only when this bus dropped an event the client still needs. Judging gaps by evictions, not by the tail's oldest seq, is required under the shared cursor space: each bus's tail is non-contiguous in the global sequence because the other bus owns the missing seqs, so comparing against tail.front() would flag false gaps. Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * feat(server): resume WatchSandbox from cursor across log and platform buses Wire resume_after_cursor into the watch producer. On a non-zero cursor, replay events strictly after it from both the log and platform buses, merge by shared cursor, and emit in order before entering the live loop. A trimmed range on either bus is an unrecoverable gap and terminates the stream with OUT_OF_RANGE carrying the requested and earliest-available cursors, distinct from recoverable lag which warns and continues. Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * test(server): cover WatchSandbox cursor resume paths Add handler-level tests for the resumable watch stream: replay strictly after the client cursor, merge log and platform events in shared-cursor order, suppress duplicates when resuming at the latest cursor, and terminate with OUT_OF_RANGE when the requested cursor has been trimmed. Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * docs(api): document WatchSandbox loss-awareness and resume Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(server): deliver watch events once and harden cursor teardown Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * feat(sdk): add loss-aware resumable watch_logs to Rust SDK client Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(server): keep watch cursors monotonic across teardown and restart Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(server): merge live watch sources by cursor before emission Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(api): bind watch cursors to a cursor space and merge tail sources Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(server): revalidate the watch cursor space after collecting replay Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * test(server): update the public RPC schema fingerprint for the string cursor Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(server): hold watch events above the publication watermark and emit the watch lag warning before its batch Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * test(server): synchronize the watch live-order test with the end of initialization Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(sdk): use canonical sandbox name in watch_logs Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * test(sdk): guard canonical-name addressing in watch_logs Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(server): fix public rpc schema Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> * fix(api): reconcile watch resume rebase Signed-off-by: John Myers <9696606+johntmyers@users.noreply.github.com> * fix(server): bound interactive relay cleanup Signed-off-by: John Myers <9696606+johntmyers@users.noreply.github.com> --------- Signed-off-by: Artem Lytvyn <alytvyn@redhat.com> Signed-off-by: John Myers <9696606+johntmyers@users.noreply.github.com> Co-authored-by: John Myers <9696606+johntmyers@users.noreply.github.com>
This commit is contained in:
Artem Lytvyn
co-authored by
John Myers
John Myers
parent
88357775ad
commit
470a34635d
@@ -7467,9 +7467,27 @@ type WatchSandboxRequest struct {
|
||||
// Filter by log source (e.g. "gateway", "sandbox"). Empty means all sources.
|
||||
LogSources []string `protobuf:"bytes,9,rep,name=log_sources,json=logSources,proto3" json:"log_sources,omitempty"`
|
||||
// Minimum log level to include (e.g. "INFO", "WARN", "ERROR"). Empty means all levels.
|
||||
LogMinLevel string `protobuf:"bytes,10,opt,name=log_min_level,json=logMinLevel,proto3" json:"log_min_level,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
LogMinLevel string `protobuf:"bytes,10,opt,name=log_min_level,json=logMinLevel,proto3" json:"log_min_level,omitempty"`
|
||||
// Resume streaming after this cursor. Empty means no cursor resume: the
|
||||
// server falls back to tail-limited replay controlled by log_tail_lines and
|
||||
// event_tail. Otherwise set it to the highest `SandboxStreamEvent.cursor`
|
||||
// already processed; the server replays only log and platform events after
|
||||
// it, merged in cursor order, before resuming live delivery. If the requested
|
||||
// cursor has already been trimmed from the server's buffer, the resume is
|
||||
// unrecoverable and the stream terminates with OUT_OF_RANGE (see
|
||||
// SandboxStreamWarning for the recoverable case).
|
||||
//
|
||||
// A cursor is bound to the cursor space that issued it. A gateway restart,
|
||||
// teardown of the sandbox's buffers, or a reconnect to a different gateway
|
||||
// replica starts a new space, and cursors from the previous one are rejected
|
||||
// with OUT_OF_RANGE rather than silently suppressing live events beneath
|
||||
// them. OUT_OF_RANGE is terminal for that cursor: restart the watch with an
|
||||
// empty resume_after_cursor, because retrying the same token fails
|
||||
// identically. A cursor this server could not have issued is rejected with
|
||||
// INVALID_ARGUMENT.
|
||||
ResumeAfterCursor string `protobuf:"bytes,12,opt,name=resume_after_cursor,json=resumeAfterCursor,proto3" json:"resume_after_cursor,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *WatchSandboxRequest) Reset() {
|
||||
@@ -7579,6 +7597,13 @@ func (x *WatchSandboxRequest) GetLogMinLevel() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *WatchSandboxRequest) GetResumeAfterCursor() string {
|
||||
if x != nil {
|
||||
return x.ResumeAfterCursor
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// One event in a sandbox watch stream.
|
||||
type SandboxStreamEvent struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
@@ -7589,7 +7614,19 @@ type SandboxStreamEvent struct {
|
||||
// *SandboxStreamEvent_Event
|
||||
// *SandboxStreamEvent_Warning
|
||||
// *SandboxStreamEvent_DraftPolicyUpdate
|
||||
Payload isSandboxStreamEvent_Payload `protobuf_oneof:"payload"`
|
||||
Payload isSandboxStreamEvent_Payload `protobuf_oneof:"payload"`
|
||||
// Opaque position in this sandbox's cursor space, shared across the resumable
|
||||
// log and platform event sources. Empty for non-resumable events (status
|
||||
// snapshots, warnings).
|
||||
//
|
||||
// Do not parse this token; its encoding is not part of the contract. The only
|
||||
// supported operation is comparing two non-empty cursors observed on the same
|
||||
// stream and keeping the greater one, then passing it as
|
||||
// WatchSandboxRequest.resume_after_cursor to resume without loss or
|
||||
// duplication. That comparison is a plain byte-wise string comparison. It is
|
||||
// well defined only within one stream: a stream never spans two cursor
|
||||
// spaces, because a reset ends it.
|
||||
Cursor string `protobuf:"bytes,6,opt,name=cursor,proto3" json:"cursor,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -7676,6 +7713,13 @@ func (x *SandboxStreamEvent) GetDraftPolicyUpdate() *DraftPolicyUpdate {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *SandboxStreamEvent) GetCursor() string {
|
||||
if x != nil {
|
||||
return x.Cursor
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type isSandboxStreamEvent_Payload interface {
|
||||
isSandboxStreamEvent_Payload()
|
||||
}
|
||||
@@ -7696,7 +7740,9 @@ type SandboxStreamEvent_Event struct {
|
||||
}
|
||||
|
||||
type SandboxStreamEvent_Warning struct {
|
||||
// Warning from the server (e.g. missed messages due to lag).
|
||||
// Recoverable warning from the server, e.g. messages dropped because a
|
||||
// broadcast receiver lagged. The stream continues after this warning; the
|
||||
// client can detect the gap from the warning itself.
|
||||
Warning *SandboxStreamWarning `protobuf:"bytes,4,opt,name=warning,proto3,oneof"`
|
||||
}
|
||||
|
||||
@@ -7811,6 +7857,11 @@ func (x *SandboxLogLine) GetFields() map[string]string {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Recoverable loss notification on a watch stream. Emitted when the server
|
||||
// skips ahead after a broadcast lag instead of terminating; the stream keeps
|
||||
// running. Cursors are opaque, so this message is the only signal that events
|
||||
// were skipped. Unrecoverable loss (a trimmed or foreign resume cursor) is
|
||||
// reported as an OUT_OF_RANGE stream status, not this message.
|
||||
type SandboxStreamWarning struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
|
||||
@@ -17976,7 +18027,7 @@ const file_openshell_proto_rawDesc = "" +
|
||||
"sandbox_id\x18\x02 \x01(\tR\tsandboxId\x12\x1a\n" +
|
||||
"\x05token\x18\x03 \x01(\tB\x04\x88\xb5\x18\x01R\x05token\x12C\n" +
|
||||
"\x0fexpiration_time\x18h \x01(\v2\x1a.google.protobuf.TimestampR\x0eexpirationTime\x12\x18\n" +
|
||||
"\arevoked\x18\x05 \x01(\bR\arevokedJ\x04\b\x04\x10\x05R\rexpires_at_ms\"\xf1\x03\n" +
|
||||
"\arevoked\x18\x05 \x01(\bR\arevokedJ\x04\b\x04\x10\x05R\rexpires_at_ms\"\xa1\x04\n" +
|
||||
"\x13WatchSandboxRequest\x12R\n" +
|
||||
"\x0fworkspace_scope\x18\v \x01(\v2).openshell.datamodel.v1.WorkspaceSelectorR\x0eworkspaceScope\x12\x18\n" +
|
||||
"\asandbox\x18\x01 \x01(\tR\asandbox\x12#\n" +
|
||||
@@ -17993,13 +18044,15 @@ const file_openshell_proto_rawDesc = "" +
|
||||
"\vlog_sources\x18\t \x03(\tR\n" +
|
||||
"logSources\x12\"\n" +
|
||||
"\rlog_min_level\x18\n" +
|
||||
" \x01(\tR\vlogMinLevelJ\x04\b\b\x10\tR\flog_since_ms\"\xcc\x02\n" +
|
||||
" \x01(\tR\vlogMinLevel\x12.\n" +
|
||||
"\x13resume_after_cursor\x18\f \x01(\tR\x11resumeAfterCursorJ\x04\b\b\x10\tR\flog_since_ms\"\xe4\x02\n" +
|
||||
"\x12SandboxStreamEvent\x121\n" +
|
||||
"\asandbox\x18\x01 \x01(\v2\x15.openshell.v1.SandboxH\x00R\asandbox\x120\n" +
|
||||
"\x03log\x18\x02 \x01(\v2\x1c.openshell.v1.SandboxLogLineH\x00R\x03log\x123\n" +
|
||||
"\x05event\x18\x03 \x01(\v2\x1b.openshell.v1.PlatformEventH\x00R\x05event\x12>\n" +
|
||||
"\awarning\x18\x04 \x01(\v2\".openshell.v1.SandboxStreamWarningH\x00R\awarning\x12Q\n" +
|
||||
"\x13draft_policy_update\x18\x05 \x01(\v2\x1f.openshell.v1.DraftPolicyUpdateH\x00R\x11draftPolicyUpdateB\t\n" +
|
||||
"\x13draft_policy_update\x18\x05 \x01(\v2\x1f.openshell.v1.DraftPolicyUpdateH\x00R\x11draftPolicyUpdate\x12\x16\n" +
|
||||
"\x06cursor\x18\x06 \x01(\tR\x06cursorB\t\n" +
|
||||
"\apayload\"\xdb\x02\n" +
|
||||
"\x0eSandboxLogLine\x12\x1d\n" +
|
||||
"\n" +
|
||||
|
||||
Reference in New Issue
Block a user