tegwick
791e419973
Some checks failed
ci / build (push) Failing after 23s
Completes the request path for FLUID-WP-0003 T01-T03 and T05-T07. The gateway resolves a revision, routes to the adapter process, and records what happened, without ever depending on the control plane to serve. Three behaviours carry tests because the architecture rests on them: Emit never blocks against a stalled sink (Blueprint 34.2), the gateway keeps serving after control-plane loss (invariant 2), and backend internals do not leak into error responses (5.7). The connector retries only idempotent methods, so a slow adapter cannot cause a hall-of-helix entry to be published twice, and breakers are per-revision so a broken candidate does not take the stable revision down with it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014KmVxhJ35tCo7rE7UnLwWu Assistant: claude-code Assistant-Model: opus Assistant-Process: 1116572@bnt-lap001 Assistant-Session: 8ba9bb93-a72a-4883-b189-2499cce5c400
101 lines
2.5 KiB
Go
101 lines
2.5 KiB
Go
package runtime
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/tegwick/fluid-core/internal/contract"
|
|
)
|
|
|
|
// stallingSink never returns, standing in for a wedged evidence store.
|
|
type stallingSink struct{ entered chan struct{} }
|
|
|
|
func (s *stallingSink) Write(ctx context.Context, _ contract.FluidTelemetry) error {
|
|
select {
|
|
case s.entered <- struct{}{}:
|
|
default:
|
|
}
|
|
<-ctx.Done()
|
|
return ctx.Err()
|
|
}
|
|
|
|
// TestEmitNeverBlocks is the test for Blueprint invariant 34.2. If it ever
|
|
// starts failing, the data plane has acquired a dependency on the observation
|
|
// plane and the isolation the whole architecture rests on is gone.
|
|
func TestEmitNeverBlocks(t *testing.T) {
|
|
sink := &stallingSink{entered: make(chan struct{}, 1)}
|
|
e := NewEmitter(sink, EmitterOptions{Buffer: 8, Workers: 1, WriteTimeout: time.Hour})
|
|
defer func() {
|
|
// The stalled worker cannot drain, so do not wait on Close.
|
|
_ = e
|
|
}()
|
|
|
|
<-func() chan struct{} {
|
|
e.Emit(contract.FluidTelemetry{ID: "warm"})
|
|
return sink.entered
|
|
}()
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
// Far more than the buffer holds, against a sink that never completes.
|
|
for i := 0; i < 10_000; i++ {
|
|
e.Emit(contract.FluidTelemetry{ID: "ev"})
|
|
}
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("Emit blocked while the sink was stalled; the request path is now coupled to telemetry")
|
|
}
|
|
|
|
if e.Stats().Dropped == 0 {
|
|
t.Error("expected drops against a stalled sink, got none")
|
|
}
|
|
}
|
|
|
|
// countingSink records deliveries.
|
|
type countingSink struct {
|
|
ch chan contract.FluidTelemetry
|
|
err error
|
|
}
|
|
|
|
func (s *countingSink) Write(_ context.Context, ev contract.FluidTelemetry) error {
|
|
if s.err != nil {
|
|
return s.err
|
|
}
|
|
s.ch <- ev
|
|
return nil
|
|
}
|
|
|
|
func TestEmitterDeliversAndDrains(t *testing.T) {
|
|
sink := &countingSink{ch: make(chan contract.FluidTelemetry, 32)}
|
|
e := NewEmitter(sink, EmitterOptions{Buffer: 32, Workers: 2})
|
|
|
|
for i := 0; i < 10; i++ {
|
|
e.Emit(contract.FluidTelemetry{ID: "ev", Kind: contract.FluidTelemetryKindRequest})
|
|
}
|
|
e.Close()
|
|
|
|
if got := len(sink.ch); got != 10 {
|
|
t.Errorf("delivered %d events, want 10", got)
|
|
}
|
|
if s := e.Stats(); s.Written != 10 || s.Dropped != 0 {
|
|
t.Errorf("stats = %+v, want 10 written and 0 dropped", s)
|
|
}
|
|
}
|
|
|
|
func TestEmitterCountsSinkFailures(t *testing.T) {
|
|
e := NewEmitter(&countingSink{err: errors.New("store down")}, EmitterOptions{Buffer: 4, Workers: 1})
|
|
for i := 0; i < 4; i++ {
|
|
e.Emit(contract.FluidTelemetry{ID: "ev"})
|
|
}
|
|
e.Close()
|
|
|
|
if s := e.Stats(); s.Failed == 0 {
|
|
t.Errorf("sink failures not counted: %+v", s)
|
|
}
|
|
}
|