From cfb269393076639c3d66bd372ee47d6753917392 Mon Sep 17 00:00:00 2001 From: cabbage Date: Wed, 2 Sep 2026 06:02:44 +0000 Subject: [PATCH] feat: add bounded session scheduling --- internal/server/session/session.go | 277 ++++++++++++++++++++++++ internal/server/session/session_test.go | 106 +++++++++ test/coverage.toml | 18 ++ 3 files changed, 401 insertions(+) create mode 100644 internal/server/session/session.go create mode 100644 internal/server/session/session_test.go diff --git a/internal/server/session/session.go b/internal/server/session/session.go new file mode 100644 index 0000000..fcd4dc4 --- /dev/null +++ b/internal/server/session/session.go @@ -0,0 +1,277 @@ +// Package session owns transport-neutral server session scheduling and fencing. +package session + +import ( + "context" + "errors" + "sync" + "time" +) + +var ( + ErrControlLaneFull = errors.New("session control lane is full") + ErrSessionClosed = errors.New("session is closed") +) + +type FrameKind uint8 + +const ( + FrameControl FrameKind = iota + 1 + FrameData +) + +type Frame struct { + Kind FrameKind + Payload []byte +} + +// WriterQueue is owned by one socket writer. Data saturation leaves the work +// durable; control saturation is a session-fatal invariant. +type WriterQueue struct { + control chan Frame + data chan Frame + controlBurst uint32 + controlRun uint32 + closed chan struct{} + closeOnce sync.Once +} + +func NewWriterQueue(controlCapacity, dataCapacity, controlBurst uint32) *WriterQueue { + if controlCapacity == 0 || dataCapacity == 0 || controlBurst == 0 { + panic("session queue capacities must be positive") + } + return &WriterQueue{ + control: make(chan Frame, controlCapacity), data: make(chan Frame, dataCapacity), + controlBurst: controlBurst, closed: make(chan struct{}), + } +} + +func (queue *WriterQueue) EnqueueControl(frame Frame) error { + if frame.Kind != FrameControl { + return ErrControlLaneFull + } + select { + case <-queue.closed: + return ErrSessionClosed + default: + } + select { + case queue.control <- frame: + return nil + default: + return ErrControlLaneFull + } +} + +// EnqueueData reports false when the bounded data lane is full. It never +// creates a fallback goroutine or buffer. +func (queue *WriterQueue) EnqueueData(frame Frame) bool { + if frame.Kind != FrameData { + return false + } + select { + case <-queue.closed: + return false + default: + } + select { + case queue.data <- frame: + return true + default: + return false + } +} + +func (queue *WriterQueue) Next(ctx context.Context) (Frame, error) { + for { + if queue.controlRun < queue.controlBurst { + select { + case frame := <-queue.control: + queue.controlRun++ + return frame, nil + default: + } + } + select { + case frame := <-queue.data: + queue.controlRun = 0 + return frame, nil + default: + } + select { + case frame := <-queue.control: + queue.controlRun++ + return frame, nil + default: + } + select { + case <-ctx.Done(): + return Frame{}, ctx.Err() + case <-queue.closed: + return Frame{}, ErrSessionClosed + case frame := <-queue.control: + queue.controlRun++ + return frame, nil + case frame := <-queue.data: + queue.controlRun = 0 + return frame, nil + } + } +} + +func (queue *WriterQueue) Close() { queue.closeOnce.Do(func() { close(queue.closed) }) } + +type HeartbeatAction uint8 + +const ( + HeartbeatNone HeartbeatAction = iota + HeartbeatPing + HeartbeatClose +) + +type Heartbeat struct { + idle time.Duration + timeout time.Duration + lastInbound time.Duration + pingSent bool +} + +func NewHeartbeat(idle, timeout, now time.Duration) *Heartbeat { + if idle <= 0 || timeout <= idle { + panic("invalid heartbeat timings") + } + return &Heartbeat{idle: idle, timeout: timeout, lastInbound: now} +} + +func (heartbeat *Heartbeat) ObserveInbound(now time.Duration) { + if now >= heartbeat.lastInbound { + heartbeat.lastInbound = now + heartbeat.pingSent = false + } +} + +func (heartbeat *Heartbeat) Check(now time.Duration) HeartbeatAction { + if now < heartbeat.lastInbound { + return HeartbeatNone + } + elapsed := now - heartbeat.lastInbound + if elapsed >= heartbeat.timeout { + return HeartbeatClose + } + if elapsed >= heartbeat.idle && !heartbeat.pingSent { + heartbeat.pingSent = true + return HeartbeatPing + } + return HeartbeatNone +} + +type DispatchLane uint8 + +const ( + DispatchNone DispatchLane = iota + DispatchRunning + DispatchQueued +) + +type CapacityShadow struct { + running, queued uint32 + maxRunning, maxQueued uint32 + shadowRunning uint32 + shadowQueued uint32 +} + +func (shadow *CapacityShadow) UpdateAdvertised(running, queued, maxRunning, maxQueued uint32) bool { + if maxRunning == 0 || maxQueued == 0 || running > maxRunning || queued > maxQueued { + return false + } + shadow.running, shadow.queued = running, queued + shadow.maxRunning, shadow.maxQueued = maxRunning, maxQueued + return true +} + +func (shadow *CapacityShadow) Reserve() DispatchLane { + if shadow.running+shadow.shadowRunning < shadow.maxRunning { + shadow.shadowRunning++ + return DispatchRunning + } + if shadow.queued+shadow.shadowQueued < shadow.maxQueued { + shadow.shadowQueued++ + return DispatchQueued + } + return DispatchNone +} + +func (shadow *CapacityShadow) Release(lane DispatchLane) bool { + switch lane { + case DispatchRunning: + if shadow.shadowRunning == 0 { + return false + } + shadow.shadowRunning-- + case DispatchQueued: + if shadow.shadowQueued == 0 { + return false + } + shadow.shadowQueued-- + default: + return false + } + return true +} + +func (shadow CapacityShadow) Pending() (running, queued uint32) { + return shadow.shadowRunning, shadow.shadowQueued +} + +type Handle struct { + ClientID string + SessionID [16]byte + Generation uint64 + Context context.Context + cancel context.CancelFunc +} + +type Registry struct { + mu sync.Mutex + clients map[string]*Handle +} + +func NewRegistry() *Registry { return &Registry{clients: make(map[string]*Handle)} } + +// Install replaces only an older in-memory handle for the same client. The +// caller must have already durably fenced it in the store. +func (registry *Registry) Install(clientID string, sessionID [16]byte, generation uint64) (*Handle, error) { + if clientID == "" || sessionID == [16]byte{} || generation == 0 { + return nil, ErrSessionClosed + } + registry.mu.Lock() + defer registry.mu.Unlock() + if old := registry.clients[clientID]; old != nil { + old.cancel() + } + ctx, cancel := context.WithCancel(context.Background()) + handle := &Handle{ClientID: clientID, SessionID: sessionID, Generation: generation, Context: ctx, cancel: cancel} + registry.clients[clientID] = handle + return handle, nil +} + +func (registry *Registry) Remove(handle *Handle) bool { + if handle == nil { + return false + } + registry.mu.Lock() + defer registry.mu.Unlock() + current := registry.clients[handle.ClientID] + if current != handle || current.Generation != handle.Generation || current.SessionID != handle.SessionID { + return false + } + delete(registry.clients, handle.ClientID) + handle.cancel() + return true +} + +func (registry *Registry) Get(clientID string) *Handle { + registry.mu.Lock() + defer registry.mu.Unlock() + return registry.clients[clientID] +} diff --git a/internal/server/session/session_test.go b/internal/server/session/session_test.go new file mode 100644 index 0000000..4bbfaa4 --- /dev/null +++ b/internal/server/session/session_test.go @@ -0,0 +1,106 @@ +package session + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestWriterQueuePriorityAndFairness_HP_SES_03(t *testing.T) { + t.Parallel() + queue := NewWriterQueue(4, 1, 2) + if !queue.EnqueueData(Frame{Kind: FrameData, Payload: []byte("data")}) { + t.Fatal("initial data enqueue failed") + } + if queue.EnqueueData(Frame{Kind: FrameData}) { + t.Fatal("full data lane accepted another frame") + } + for _, value := range []string{"control-1", "control-2", "control-3"} { + if err := queue.EnqueueControl(Frame{Kind: FrameControl, Payload: []byte(value)}); err != nil { + t.Fatal(err) + } + } + ctx := context.Background() + for index, want := range []string{"control-1", "control-2", "data", "control-3"} { + frame, err := queue.Next(ctx) + if err != nil || string(frame.Payload) != want { + t.Fatalf("frame %d = (%q, %v), want %q", index, frame.Payload, err, want) + } + } + if err := queue.EnqueueControl(Frame{Kind: FrameData}); !errors.Is(err, ErrControlLaneFull) { + t.Fatalf("wrong control frame error = %v", err) + } + queue.Close() + if _, err := queue.Next(context.Background()); !errors.Is(err, ErrSessionClosed) { + t.Fatalf("closed queue Next error = %v", err) + } +} + +func TestHeartbeatBoundaries_HP_SES_04(t *testing.T) { + t.Parallel() + heartbeat := NewHeartbeat(10*time.Second, 30*time.Second, 0) + if action := heartbeat.Check(9 * time.Second); action != HeartbeatNone { + t.Fatalf("9s action = %d", action) + } + if action := heartbeat.Check(10 * time.Second); action != HeartbeatPing { + t.Fatalf("10s action = %d", action) + } + if action := heartbeat.Check(29 * time.Second); action != HeartbeatNone { + t.Fatalf("29s action = %d", action) + } + if action := heartbeat.Check(30 * time.Second); action != HeartbeatClose { + t.Fatalf("30s action = %d", action) + } + heartbeat.ObserveInbound(31 * time.Second) + if action := heartbeat.Check(41 * time.Second); action != HeartbeatPing { + t.Fatalf("inbound reset action = %d", action) + } + if action := heartbeat.Check(1 * time.Second); action != HeartbeatNone { + t.Fatalf("clock rollback action = %d", action) + } +} + +func TestCapacityShadowAndRegistry_BH_SES_03(t *testing.T) { + t.Parallel() + var shadow CapacityShadow + if shadow.UpdateAdvertised(0, 0, 0, 1) { + t.Fatal("invalid capacity accepted") + } + if !shadow.UpdateAdvertised(0, 0, 1, 1) { + t.Fatal("valid capacity rejected") + } + if got := shadow.Reserve(); got != DispatchRunning { + t.Fatalf("first lane = %d", got) + } + if got := shadow.Reserve(); got != DispatchQueued { + t.Fatalf("second lane = %d", got) + } + if got := shadow.Reserve(); got != DispatchNone { + t.Fatalf("oversubscription lane = %d", got) + } + if !shadow.Release(DispatchRunning) || shadow.Release(DispatchRunning) { + t.Fatal("shadow release accounting is wrong") + } + + registry := NewRegistry() + first, err := registry.Install("client-a", [16]byte{1}, 1) + if err != nil { + t.Fatal(err) + } + second, err := registry.Install("client-a", [16]byte{2}, 2) + if err != nil { + t.Fatal(err) + } + select { + case <-first.Context.Done(): + default: + t.Fatal("replacement did not fence old handle") + } + if registry.Remove(first) || registry.Get("client-a") != second { + t.Fatal("stale handle removed current session") + } + if !registry.Remove(second) { + t.Fatal("current handle removal failed") + } +} diff --git a/test/coverage.toml b/test/coverage.toml index ce9a129..c3cde0f 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -357,3 +357,21 @@ id = "BH-SES-02" layer = "integration" status = "implemented" tests = ["test/integration/store/store_integration_test.go:TestTakeoverBoundariesAndClosedSession_BH_SES_02"] + +[[requirements]] +id = "HP-SES-03" +layer = "unit" +status = "implemented" +tests = ["internal/server/session/session_test.go:TestWriterQueuePriorityAndFairness_HP_SES_03"] + +[[requirements]] +id = "HP-SES-04" +layer = "unit" +status = "implemented" +tests = ["internal/server/session/session_test.go:TestHeartbeatBoundaries_HP_SES_04"] + +[[requirements]] +id = "BH-SES-03" +layer = "unit" +status = "implemented" +tests = ["internal/server/session/session_test.go:TestCapacityShadowAndRegistry_BH_SES_03"]