feat(execd): extract stream buffer (sbuf) package for ring buffer queue

Author: Pangjiping

components/execd/pkg/sbuf/config.go

@@ -0,0 +1,37 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sbuf
+
+// Config controls per-stream bounds and append policy.
+type Config struct {
+	// MaxEvents is the maximum number of events retained per stream. Oldest events are dropped when exceeded.
+	// Zero defaults to DefaultMaxEvents.
+	MaxEvents int
+	// MaxBytes is the approximate upper bound on total payload bytes per stream (sum of len(Payload)).
+	// Oldest events are dropped until under the limit. Zero means no byte limit.
+	MaxBytes int64
+	// StrictMonotonic rejects Append when eid <= last eid for that stream. Recommended for execd SSE eids.
+	StrictMonotonic bool
+}
+
+const DefaultMaxEvents = 1024
+
+func (c *Config) normalized() Config {
+	out := *c
+	if out.MaxEvents <= 0 {
+		out.MaxEvents = DefaultMaxEvents
+	}
+	return out
+}

components/execd/pkg/sbuf/errors.go

@@ -0,0 +1,23 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sbuf
+
+import "errors"
+
+var (
+	// ErrOutOfOrder is returned when StrictMonotonic is enabled and eid is not greater than the last appended eid.
+	ErrOutOfOrder    = errors.New("sbuf: eid out of order for stream")
+	ErrEmptyStreamID = errors.New("sbuf: empty stream id")
+)

components/execd/pkg/sbuf/event.go

@@ -0,0 +1,21 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sbuf
+
+// Event is one stored record (e.g. a single SSE JSON line body). Payload is owned by the buffer after Append.
+type Event struct {
+	EID     int64
+	Payload []byte
+}

components/execd/pkg/sbuf/ring.go

@@ -0,0 +1,91 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sbuf
+
+// ring is a FIFO queue with a fixed max length; push drops oldest when full.
+type ring struct {
+	maxLen int
+	slots  []eventSlot
+	head   int
+	n      int
+	bytes  int64
+}
+
+type eventSlot struct {
+	eid     int64
+	payload []byte
+}
+
+func newRing(maxLen int) *ring {
+	if maxLen < 1 {
+		maxLen = 1
+	}
+	return &ring{
+		maxLen: maxLen,
+		slots:  make([]eventSlot, maxLen),
+	}
+}
+
+func (r *ring) push(eid int64, payload []byte, maxBytes int64) {
+	pld := append([]byte(nil), payload...)
+	size := int64(len(pld))
+
+	if r.n == r.maxLen {
+		r.evictHead()
+	}
+	idx := (r.head + r.n) % r.maxLen
+	r.slots[idx] = eventSlot{eid: eid, payload: pld}
+	r.n++
+	r.bytes += size
+
+	if maxBytes > 0 {
+		for r.bytes > maxBytes && r.n > 0 {
+			r.evictHead()
+		}
+	}
+}
+
+func (r *ring) evictHead() {
+	if r.n == 0 {
+		return
+	}
+	old := r.slots[r.head]
+	r.bytes -= int64(len(old.payload))
+	r.slots[r.head] = eventSlot{}
+	r.head = (r.head + 1) % r.maxLen
+	r.n--
+}
+
+func (r *ring) iterAfter(afterEid int64, fn func(eid int64, payload []byte)) {
+	for i := range r.n {
+		idx := (r.head + i) % r.maxLen
+		s := r.slots[idx]
+		if s.eid > afterEid {
+			fn(s.eid, s.payload)
+		}
+	}
+}
+
+// snapshotAfter returns a copy slice for safe iteration outside the ring lock.
+func (r *ring) snapshotAfter(afterEid int64) []Event {
+	var out []Event
+	r.iterAfter(afterEid, func(eid int64, payload []byte) {
+		out = append(out, Event{
+			EID:     eid,
+			Payload: append([]byte(nil), payload...),
+		})
+	})
+	return out
+}

components/execd/pkg/sbuf/store.go

@@ -0,0 +1,108 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package sbuf provides bounded, per-stream FIFO buffers for SSE (or similar) events keyed by eid,
+// used to serve disconnect resume (catch-up by event id).
+// It is storage-only: callers assign eids and decide when to delete a stream.
+package sbuf
+
+import (
+	"sync"
+)
+
+// Store holds bounded event rings keyed by caller-defined stream IDs (e.g. command execution id).
+type Store struct {
+	cfg     Config
+	mu      sync.Mutex
+	streams map[string]*streamBuf
+}
+
+type streamBuf struct {
+	mu       sync.Mutex
+	lastEid  int64
+	ring     *ring
+	maxBytes int64
+}
+
+// NewStore creates an empty store. cfg is copied after normalization.
+func NewStore(cfg Config) *Store {
+	cfg = cfg.normalized()
+	return &Store{
+		cfg:     cfg,
+		streams: make(map[string]*streamBuf),
+	}
+}
+
+// Append adds one event to the stream's ring. Payload is copied.
+// With StrictMonotonic, returns ErrOutOfOrder if eid <= previous eid for this stream.
+func (s *Store) Append(streamID string, eid int64, payload []byte) error {
+	if streamID == "" {
+		return ErrEmptyStreamID
+	}
+	sb := s.getOrCreate(streamID)
+	sb.mu.Lock()
+	defer sb.mu.Unlock()
+
+	if s.cfg.StrictMonotonic {
+		if eid <= sb.lastEid {
+			return ErrOutOfOrder
+		}
+	}
+	sb.lastEid = eid
+	sb.ring.push(eid, payload, sb.maxBytes)
+	return nil
+}
+
+func (s *Store) getOrCreate(streamID string) *streamBuf {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if sb, ok := s.streams[streamID]; ok {
+		return sb
+	}
+	sb := &streamBuf{
+		ring:     newRing(s.cfg.MaxEvents),
+		maxBytes: s.cfg.MaxBytes,
+	}
+	s.streams[streamID] = sb
+	return sb
+}
+
+// EventsAfter returns a snapshot of events with EID > afterEid in order.
+// If the stream does not exist, ok is false and events is nil.
+func (s *Store) EventsAfter(streamID string, afterEid int64) (events []Event, ok bool) {
+	s.mu.Lock()
+	sb, found := s.streams[streamID]
+	s.mu.Unlock()
+	if !found {
+		return nil, false
+	}
+	sb.mu.Lock()
+	defer sb.mu.Unlock()
+	return sb.ring.snapshotAfter(afterEid), true
+}
+
+// Delete removes a stream buffer. No-op if missing.
+func (s *Store) Delete(streamID string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.streams, streamID)
+}
+
+// Has reports whether a stream currently exists.
+func (s *Store) Has(streamID string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	_, ok := s.streams[streamID]
+	return ok
+}