diff --git a/internal/runtime/connector.go b/internal/runtime/connector.go new file mode 100644 index 0000000..33256eb --- /dev/null +++ b/internal/runtime/connector.go @@ -0,0 +1,248 @@ +package runtime + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Connector calls the adapter process serving a revision. +// +// ArchitectureBlueprint.md section 5.6 puts protocol and dependency detail +// behind this boundary. Under out-of-process attachment (ADR-0002) the adapter +// is reached over HTTP, which is what lets it be written in any language. +type Connector struct { + client *http.Client + breakers sync.Map // contract.RevisionID -> *breaker + now func() time.Time +} + +// NewConnector returns a connector with sane transport defaults. +func NewConnector() *Connector { + return &Connector{ + client: &http.Client{ + // No client-level timeout: per-revision timeouts come from the + // descriptor and are applied through the request context, so one + // slow revision cannot impose its budget on another. + Transport: &http.Transport{ + MaxIdleConns: 100, + MaxIdleConnsPerHost: 16, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 5 * time.Second, + ExpectContinueTimeout: time.Second, + }, + }, + now: time.Now, + } +} + +// UpstreamError describes a failed adapter call, classified for the response +// policy and for telemetry. +type UpstreamError struct { + Kind ErrorKind + Err error +} + +func (e *UpstreamError) Error() string { return fmt.Sprintf("%s: %v", e.Kind, e.Err) } +func (e *UpstreamError) Unwrap() error { return e.Err } + +// ErrCircuitOpen is returned while a revision's breaker is open. +var ErrCircuitOpen = errors.New("circuit open") + +// Call forwards a request to the adapter serving rev. +func (c *Connector) Call(ctx context.Context, rev contract.Revision, r *http.Request, body io.Reader) (*http.Response, error) { + b := c.breakerFor(rev) + if !b.allow(c.now()) { + return nil, &UpstreamError{Kind: ErrorUnavailable, Err: ErrCircuitOpen} + } + + timeout := 5 * time.Second + if rev.Runtime.TimeoutMS != nil && *rev.Runtime.TimeoutMS > 0 { + timeout = time.Duration(*rev.Runtime.TimeoutMS) * time.Millisecond + } + + attempts := 1 + var backoff time.Duration + if rev.Runtime.Retry != nil { + if rev.Runtime.Retry.MaxAttempts != nil && *rev.Runtime.Retry.MaxAttempts > 1 { + attempts = int(*rev.Runtime.Retry.MaxAttempts) + } + if rev.Runtime.Retry.BackoffMS != nil { + backoff = time.Duration(*rev.Runtime.Retry.BackoffMS) * time.Millisecond + } + } + + // Only idempotent methods are retried. Replaying a POST because an adapter + // was slow would publish the same hall-of-helix entry twice. + if !idempotent(r.Method) { + attempts = 1 + } + + var lastErr error + for attempt := 0; attempt < attempts; attempt++ { + if attempt > 0 { + select { + case <-ctx.Done(): + return nil, &UpstreamError{Kind: ErrorTimeout, Err: ctx.Err()} + case <-time.After(backoff): + } + } + + resp, err := c.attempt(ctx, rev, r, body, timeout) + if err == nil { + b.record(true, c.now(), rev) + return resp, nil + } + lastErr = err + + // A retry needs a rewindable body; without one, one attempt is all + // there is. + if body != nil { + if seeker, ok := body.(io.Seeker); ok { + if _, serr := seeker.Seek(0, io.SeekStart); serr != nil { + break + } + } else { + break + } + } + } + + b.record(false, c.now(), rev) + return nil, lastErr +} + +func (c *Connector) attempt(ctx context.Context, rev contract.Revision, r *http.Request, body io.Reader, timeout time.Duration) (*http.Response, error) { + callCtx, cancel := context.WithTimeout(ctx, timeout) + + target := strings.TrimSuffix(rev.Runtime.Upstream, "/") + r.URL.RequestURI() + req, err := http.NewRequestWithContext(callCtx, r.Method, target, body) + if err != nil { + cancel() + return nil, &UpstreamError{Kind: ErrorUnavailable, Err: err} + } + + // Forward consumer headers, minus hop-by-hop ones. + for k, vs := range r.Header { + if hopByHop[strings.ToLower(k)] { + continue + } + for _, v := range vs { + req.Header.Add(k, v) + } + } + req.Header.Set("X-FLUID-Revision", string(rev.ID)) + req.Header.Set("X-FLUID-Interface", string(rev.Interface)) + + resp, err := c.client.Do(req) + if err != nil { + cancel() + kind := ErrorUnavailable + var netErr net.Error + if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()) { + kind = ErrorTimeout + } + return nil, &UpstreamError{Kind: kind, Err: err} + } + + // The body is still streaming, so cancellation has to outlive this call. + resp.Body = &cancelOnClose{ReadCloser: resp.Body, cancel: cancel} + return resp, nil +} + +func idempotent(method string) bool { + switch method { + case http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodPut, http.MethodDelete: + return true + } + return false +} + +var hopByHop = map[string]bool{ + "connection": true, "keep-alive": true, "proxy-authenticate": true, + "proxy-authorization": true, "te": true, "trailer": true, + "transfer-encoding": true, "upgrade": true, +} + +type cancelOnClose struct { + io.ReadCloser + cancel context.CancelFunc + once sync.Once +} + +func (c *cancelOnClose) Close() error { + err := c.ReadCloser.Close() + c.once.Do(c.cancel) + return err +} + +// breaker is a per-revision circuit breaker. +// +// It is per-revision rather than per-host because two revisions frequently +// share a host: an adapter that is broken at R-3 should not take R-2 down +// with it. +type breaker struct { + mu sync.Mutex + failures int64 + threshold int64 + openUntil time.Time + resetAfter time.Duration + probeAllowed bool +} + +func (c *Connector) breakerFor(rev contract.Revision) *breaker { + if v, ok := c.breakers.Load(rev.ID); ok { + return v.(*breaker) + } + b := &breaker{threshold: 5, resetAfter: 30 * time.Second} + if cb := rev.Runtime.CircuitBreaker; cb != nil { + if cb.FailureThreshold != nil && *cb.FailureThreshold > 0 { + b.threshold = *cb.FailureThreshold + } + if cb.ResetAfterMS != nil && *cb.ResetAfterMS > 0 { + b.resetAfter = time.Duration(*cb.ResetAfterMS) * time.Millisecond + } + } + actual, _ := c.breakers.LoadOrStore(rev.ID, b) + return actual.(*breaker) +} + +func (b *breaker) allow(now time.Time) bool { + b.mu.Lock() + defer b.mu.Unlock() + if b.openUntil.IsZero() || now.After(b.openUntil) { + if !b.openUntil.IsZero() { + // Half-open: let exactly one request through to test recovery. + if b.probeAllowed { + return false + } + b.probeAllowed = true + } + return true + } + return false +} + +func (b *breaker) record(success bool, now time.Time, rev contract.Revision) { + b.mu.Lock() + defer b.mu.Unlock() + if success { + b.failures = 0 + b.openUntil = time.Time{} + b.probeAllowed = false + return + } + b.failures++ + if b.failures >= b.threshold { + b.openUntil = now.Add(b.resetAfter) + b.probeAllowed = false + } +} diff --git a/internal/runtime/gateway.go b/internal/runtime/gateway.go new file mode 100644 index 0000000..2ed1089 --- /dev/null +++ b/internal/runtime/gateway.go @@ -0,0 +1,284 @@ +package runtime + +import ( + "bytes" + "crypto/rand" + "encoding/hex" + "errors" + "io" + "net/http" + "strings" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// CohortResolver assigns an inbound request to a consumer cohort. +// +// Cohorts should be coarse (FluidAPIStandards.md section 14): granular enough +// to compare populations, never more specific than the analysis requires. +type CohortResolver interface { + Cohort(*http.Request) (contract.CohortID, string) +} + +// StaticCohort assigns everything to one cohort. Useful before cohort analysis +// exists, and for interfaces with a single kind of consumer. +type StaticCohort contract.CohortID + +// Cohort implements CohortResolver. +func (s StaticCohort) Cohort(r *http.Request) (contract.CohortID, string) { + return contract.CohortID(s), consumerRef(r) +} + +// HeaderCohort reads the cohort from a request header, falling back to a +// default when absent or unrecognized. +type HeaderCohort struct { + Header string + Known map[string]contract.CohortID + Default contract.CohortID +} + +// Cohort implements CohortResolver. +func (h HeaderCohort) Cohort(r *http.Request) (contract.CohortID, string) { + if v := r.Header.Get(h.Header); v != "" { + if c, ok := h.Known[v]; ok { + return c, consumerRef(r) + } + } + return h.Default, consumerRef(r) +} + +// consumerRef extracts a stable, pseudonymous consumer identity. +// +// It must never be a raw end-user identifier: the telemetry envelope carries +// this value into the evidence store, and Blueprint section 6.2 requires +// pseudonymization there. +func consumerRef(r *http.Request) string { + if v := r.Header.Get("X-FLUID-Consumer"); v != "" { + return v + } + return "" +} + +// GatewayOptions configures the data plane. +type GatewayOptions struct { + Interface contract.InterfaceID + Registry *Registry + Resolver *Resolver + Connector *Connector + Emitter *Emitter + Cohorts CohortResolver + Response ResponsePolicy + + // MaxBodyBytes bounds request size. Zero applies a 1 MiB default. + MaxBodyBytes int64 + // Validator, when set, checks requests against the revision contract. + Validator ContractValidator +} + +// ContractValidator checks a request against a revision's declared contract. +// +// It is an interface rather than a concrete OpenAPI implementation because +// FluidAPIStandards.md section 4 admits several contract forms, and the gateway +// should not know which one an interface chose. +type ContractValidator interface { + Validate(rev contract.Revision, r *http.Request, body []byte) error +} + +// ValidationError reports a contract violation. +type ValidationError struct { + Field string + Message string +} + +func (e *ValidationError) Error() string { return e.Message } + +// Gateway is the deterministic entry point for interface traffic. +// +// It terminates transport, assigns correlation, resolves and routes a revision, +// calls the adapter, and emits telemetry. It does not interpret semantics: +// Blueprint section 5.1 forbids the gateway from inventing them, and section +// 48.1 names an LLM in the request path as an anti-pattern. +type Gateway struct { + opts GatewayOptions + now func() time.Time +} + +// NewGateway returns a gateway. Registry, Resolver and Connector are required. +func NewGateway(opts GatewayOptions) (*Gateway, error) { + if opts.Registry == nil || opts.Resolver == nil || opts.Connector == nil { + return nil, errors.New("gateway requires a registry, resolver and connector") + } + if opts.Cohorts == nil { + opts.Cohorts = StaticCohort("unclassified") + } + if opts.MaxBodyBytes <= 0 { + opts.MaxBodyBytes = 1 << 20 + } + return &Gateway{opts: opts, now: time.Now}, nil +} + +func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { + started := g.now() + correlation := correlationID(r) + w.Header().Set("X-FLUID-Correlation", correlation) + + cohort, consumer := g.opts.Cohorts.Cohort(r) + + body, err := io.ReadAll(io.LimitReader(r.Body, g.opts.MaxBodyBytes+1)) + if err != nil { + g.fail(w, r, correlation, cohort, "", ErrorValidation, "could not read request body", "", started) + return + } + if int64(len(body)) > g.opts.MaxBodyBytes { + g.fail(w, r, correlation, cohort, "", ErrorValidation, "request body exceeds the configured limit", "", started) + return + } + + req := Request{ + ExplicitRevision: contract.RevisionID(r.Header.Get("X-FLUID-Revision")), + BoundRevision: contract.RevisionID(r.Header.Get("X-FLUID-Bound-Revision")), + Cohort: cohort, + Tenant: r.Header.Get("X-FLUID-Tenant"), + ConsumerRef: consumer, + CorrelationID: correlation, + } + + resolution, err := g.opts.Resolver.Resolve(req) + if err != nil { + kind := ErrorUnavailable + message := "no revision is currently able to serve this request" + if errors.Is(err, ErrRevisionNotRoutable) || errors.Is(err, ErrUnknownRevision) { + kind = ErrorValidation + message = "the requested revision is not available" + } + g.fail(w, r, correlation, cohort, "", kind, message, "", started) + return + } + + rev, err := g.opts.Registry.Revision(resolution.Revision) + if err != nil { + g.fail(w, r, correlation, cohort, resolution.Revision, ErrorUnavailable, + "the resolved revision is not published", "", started) + return + } + + if g.opts.Validator != nil { + if verr := g.opts.Validator.Validate(rev, r, body); verr != nil { + field := "" + var ve *ValidationError + if errors.As(verr, &ve) { + field = ve.Field + } + g.failWith(w, r, correlation, cohort, rev.ID, ErrorValidation, verr.Error(), field, started, &resolution) + return + } + } + + w.Header().Set("X-FLUID-Revision", string(rev.ID)) + + resp, err := g.opts.Connector.Call(r.Context(), rev, r, bytes.NewReader(body)) + if err != nil { + kind := ErrorUnavailable + var ue *UpstreamError + if errors.As(err, &ue) { + kind = ue.Kind + } + g.failWith(w, r, correlation, cohort, rev.ID, kind, + "the interface could not complete this request", "", started, &resolution) + return + } + defer resp.Body.Close() + + for k, vs := range resp.Header { + if hopByHop[strings.ToLower(k)] { + continue + } + for _, v := range vs { + w.Header().Add(k, v) + } + } + w.WriteHeader(resp.StatusCode) + written, _ := io.Copy(w, resp.Body) + + g.emitRequest(r, correlation, cohort, rev.ID, &resolution, resp.StatusCode, written, int64(len(body)), started, nil) +} + +func (g *Gateway) fail(w http.ResponseWriter, r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, kind ErrorKind, msg, field string, started time.Time) { + g.failWith(w, r, correlation, cohort, rev, kind, msg, field, started, nil) +} + +func (g *Gateway) failWith(w http.ResponseWriter, r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, kind ErrorKind, msg, field string, started time.Time, res *Resolution) { + g.opts.Response.WriteError(w, kind, correlation, msg, rev, field) + status := statusFor[kind] + if status == 0 { + status = http.StatusInternalServerError + } + g.emitRequest(r, correlation, cohort, rev, res, status, 0, 0, started, &kind) +} + +// emitRequest records what happened. Errors are evidence, not noise: +// FluidAPIStandards.md principle 3 treats them as product signals. +func (g *Gateway) emitRequest(r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, res *Resolution, status int, respBytes, reqBytes int64, started time.Time, errKind *ErrorKind) { + if g.opts.Emitter == nil { + return + } + + latency := float64(g.now().Sub(started).Microseconds()) / 1000.0 + statusCode := int64(status) + ev := contract.FluidTelemetry{ + SchemaVersion: "0.1", + ID: newID("tl-"), + OccurredAt: g.now().UTC(), + InterfaceID: g.opts.Interface, + Kind: contract.FluidTelemetryKindRequest, + CorrelationID: correlation, + ConsumerRef: r.Header.Get("X-FLUID-Consumer"), + Cohort: &cohort, + Revision: &rev, + Request: &contract.FluidTelemetryRequest{ + Route: r.URL.Path, + Method: r.Method, + Status: &statusCode, + LatencyMS: &latency, + RequestBytes: &reqBytes, + ResponseBytes: &respBytes, + }, + Redaction: &contract.FluidTelemetryRedaction{Applied: false}, + } + + if res != nil { + ev.Resolution = &contract.FluidTelemetryResolution{ + Reason: res.Reason, + PolicyGeneration: &res.PolicyGeneration, + } + ev.Experiment = res.Experiment + } + + if errKind != nil { + ev.Kind = contract.FluidTelemetryKindError + class := contract.FluidTelemetryErrorClass(*errKind) + if class.Valid() { + ev.Error = &contract.FluidTelemetryError{Class: class} + } + } + + g.opts.Emitter.Emit(ev) +} + +// correlationID reuses an inbound correlation reference when the consumer +// supplied one, so a call chain stays linked across services. +func correlationID(r *http.Request) string { + if v := r.Header.Get("X-FLUID-Correlation"); v != "" { + return v + } + return newID("c-") +} + +func newID(prefix string) string { + var b [12]byte + if _, err := rand.Read(b[:]); err != nil { + return prefix + "0" + } + return prefix + hex.EncodeToString(b[:]) +} diff --git a/internal/runtime/gateway_test.go b/internal/runtime/gateway_test.go new file mode 100644 index 0000000..7042310 --- /dev/null +++ b/internal/runtime/gateway_test.go @@ -0,0 +1,215 @@ +package runtime + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// recordingSink keeps every event for assertions. +type recordingSink struct { + mu sync.Mutex + events []contract.FluidTelemetry +} + +func (s *recordingSink) Write(_ context.Context, ev contract.FluidTelemetry) error { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, ev) + return nil +} + +func (s *recordingSink) all() []contract.FluidTelemetry { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]contract.FluidTelemetry, len(s.events)) + copy(out, s.events) + return out +} + +// harness wires a gateway in front of a stub adapter. +type harness struct { + gateway *Gateway + sink *recordingSink + emitter *Emitter + reg *Registry + adapter *httptest.Server +} + +func newHarness(t *testing.T, handler http.HandlerFunc) *harness { + t.Helper() + + adapter := httptest.NewServer(handler) + t.Cleanup(adapter.Close) + + reg := NewRegistry(testInterface) + d := descriptor("R-1", contract.RevisionStateStable) + d.Runtime.Upstream = adapter.URL + if err := reg.PutRevision(d); err != nil { + t.Fatal(err) + } + if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil { + t.Fatal(err) + } + + sink := &recordingSink{} + emitter := NewEmitter(sink, EmitterOptions{Buffer: 64, Workers: 1}) + t.Cleanup(emitter.Close) + + gw, err := NewGateway(GatewayOptions{ + Interface: testInterface, + Registry: reg, + Resolver: NewResolver(reg, true), + Connector: NewConnector(), + Emitter: emitter, + Cohorts: StaticCohort("publishing-jobs"), + Response: ResponsePolicy{FeedbackPath: "/v1/feedback"}, + }) + if err != nil { + t.Fatal(err) + } + + return &harness{gateway: gw, sink: sink, emitter: emitter, reg: reg, adapter: adapter} +} + +func TestGatewayForwardsAndRecords(t *testing.T) { + h := newHarness(t, func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("X-FLUID-Revision"); got != "R-1" { + t.Errorf("adapter saw revision %q, want R-1", got) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"published":true}`)) + }) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/v1/hall-entries", strings.NewReader(`{"id":"e-1"}`)) + req.Header.Set("X-FLUID-Consumer", "hall-publisher") + h.gateway.ServeHTTP(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d, want 201; body %s", rec.Code, rec.Body.String()) + } + if rec.Header().Get("X-FLUID-Revision") != "R-1" { + t.Error("response does not name the revision that served it") + } + if rec.Header().Get("X-FLUID-Correlation") == "" { + t.Error("response carries no correlation reference") + } + + h.emitter.Close() + events := h.sink.all() + if len(events) != 1 { + t.Fatalf("emitted %d events, want 1", len(events)) + } + ev := events[0] + if ev.Kind != contract.FluidTelemetryKindRequest { + t.Errorf("kind = %s", ev.Kind) + } + if ev.Resolution == nil || ev.Resolution.Reason != contract.FluidTelemetryResolutionReasonStableDefault { + t.Errorf("resolution reason not recorded: %+v", ev.Resolution) + } + if ev.Revision == nil || *ev.Revision != "R-1" { + t.Error("revision not recorded on the event") + } + if ev.Request == nil || ev.Request.Route != "/v1/hall-entries" { + t.Errorf("request detail missing: %+v", ev.Request) + } +} + +func TestGatewaySurfacesBackendFailureWithoutLeaking(t *testing.T) { + h := newHarness(t, func(w http.ResponseWriter, r *http.Request) { + panic("adapter exploded: dsn=postgres://user:hunter2@db.internal/prod") + }) + // The stub's panic is handled by httptest's server, which closes the + // connection; the connector sees a transport failure. + h.adapter.Config.ErrorLog = nil + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v1/hall-entries/e-1", nil) + h.gateway.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadGateway && rec.Code != http.StatusGatewayTimeout { + t.Fatalf("status = %d, want a gateway error", rec.Code) + } + + var body ErrorBody + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("error body is not JSON: %v", err) + } + if strings.Contains(rec.Body.String(), "hunter2") || strings.Contains(rec.Body.String(), "postgres://") { + t.Error("backend internals leaked into the error response") + } + if body.Correlation == "" { + t.Error("error response carries no correlation reference") + } + if body.Feedback != "/v1/feedback" { + t.Error("error response does not point at the feedback endpoint") + } + + h.emitter.Close() + events := h.sink.all() + if len(events) != 1 || events[0].Kind != contract.FluidTelemetryKindError { + t.Fatalf("failure was not recorded as error telemetry: %+v", events) + } +} + +// TestGatewayServesWithoutControlPlane is the Blueprint invariant 2 check: +// evolution can stop without stopping the API. Here the control plane is +// represented by the registry's ability to accept updates; the gateway must +// keep serving from what it already holds. +func TestGatewayServesWithoutControlPlane(t *testing.T) { + h := newHarness(t, func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`ok`)) + }) + + // Simulate control-plane loss: no further policies or descriptors arrive, + // and telemetry delivery is dead. + h.emitter.Close() + + for i := 0; i < 5; i++ { + rec := httptest.NewRecorder() + h.gateway.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/v1/hall-entries", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("request %d failed with %d after control-plane loss", i, rec.Code) + } + } +} + +func TestGatewayRejectsOversizedBody(t *testing.T) { + h := newHarness(t, func(w http.ResponseWriter, r *http.Request) { + t.Error("adapter should not have been reached") + }) + h.gateway.opts.MaxBodyBytes = 16 + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/v1/hall-entries", strings.NewReader(strings.Repeat("x", 64))) + h.gateway.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400", rec.Code) + } +} + +func TestGatewayPropagatesCorrelation(t *testing.T) { + h := newHarness(t, func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("X-FLUID-Correlation"); got != "c-upstream" { + t.Errorf("adapter saw correlation %q, want it forwarded", got) + } + }) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v1/hall-entries", nil) + req.Header.Set("X-FLUID-Correlation", "c-upstream") + h.gateway.ServeHTTP(rec, req) + + if rec.Header().Get("X-FLUID-Correlation") != "c-upstream" { + t.Error("inbound correlation was not reused") + } +} diff --git a/internal/runtime/response.go b/internal/runtime/response.go new file mode 100644 index 0000000..b0ff49c --- /dev/null +++ b/internal/runtime/response.go @@ -0,0 +1,91 @@ +package runtime + +import ( + "encoding/json" + "net/http" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// ErrorKind classifies a response the interface produced rather than the +// backend. These map onto the telemetry error classes so that a response and +// the evidence it generates cannot disagree. +type ErrorKind string + +const ( + ErrorValidation ErrorKind = "validation" + ErrorUnknownPath ErrorKind = "unknown_path" + ErrorUnsupported ErrorKind = "unsupported_parameter" + ErrorAuthorization ErrorKind = "authorization" + ErrorUnavailable ErrorKind = "backend_failure" + ErrorTimeout ErrorKind = "timeout" + ErrorPolicy ErrorKind = "policy_rejection" + ErrorNoCapability ErrorKind = "missing_capability" +) + +// ErrorBody is the interface's error representation. +// +// ArchitectureBlueprint.md section 5.7: errors should carry enough structured +// information to improve observability without leaking backend detail, and may +// include a correlation reference that lets downstream analysis link the +// response a consumer saw to the pressure it generated. +type ErrorBody struct { + Kind ErrorKind `json:"kind"` + // Message is written for the consumer and must stay free of backend + // internals: hostnames, stack traces, driver errors, upstream payloads. + Message string `json:"message"` + // Correlation is the reference a consumer can quote back, and the key that + // ties this response to its telemetry. + Correlation string `json:"correlation"` + // Revision tells the consumer which contract answered. Without it a client + // debugging an unexpected response has no way to know what it was talking to. + Revision contract.RevisionID `json:"revision,omitempty"` + // Field names the offending input for validation failures. + Field string `json:"field,omitempty"` + // Feedback points at the explicit-feedback endpoint. Turning a dead end into + // an invitation is the cheapest pressure signal the interface can collect. + Feedback string `json:"feedback,omitempty"` +} + +// statusFor maps an error kind to its HTTP status. +var statusFor = map[ErrorKind]int{ + ErrorValidation: http.StatusBadRequest, + ErrorUnknownPath: http.StatusNotFound, + ErrorUnsupported: http.StatusBadRequest, + ErrorAuthorization: http.StatusForbidden, + ErrorUnavailable: http.StatusBadGateway, + ErrorTimeout: http.StatusGatewayTimeout, + ErrorPolicy: http.StatusForbidden, + ErrorNoCapability: http.StatusNotImplemented, +} + +// ResponsePolicy renders interface errors consistently. +type ResponsePolicy struct { + // FeedbackPath, when set, is advertised on every error. + FeedbackPath string +} + +// WriteError renders an error response. +func (p ResponsePolicy) WriteError(w http.ResponseWriter, kind ErrorKind, correlation, message string, rev contract.RevisionID, field string) { + status, ok := statusFor[kind] + if !ok { + status = http.StatusInternalServerError + } + + body := ErrorBody{ + Kind: kind, + Message: message, + Correlation: correlation, + Revision: rev, + Field: field, + Feedback: p.FeedbackPath, + } + + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-FLUID-Correlation", correlation) + if rev != "" { + w.Header().Set("X-FLUID-Revision", string(rev)) + } + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} diff --git a/internal/runtime/telemetry.go b/internal/runtime/telemetry.go new file mode 100644 index 0000000..974de8f --- /dev/null +++ b/internal/runtime/telemetry.go @@ -0,0 +1,143 @@ +package runtime + +import ( + "context" + "sync" + "sync/atomic" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Sink receives normalized telemetry. Implementations may block; the emitter +// exists precisely so that blocking never reaches a request. +type Sink interface { + Write(context.Context, contract.FluidTelemetry) error +} + +// Emitter buffers telemetry and delivers it out of band. +// +// ArchitectureBlueprint.md section 34.2 is a hard invariant: telemetry +// backpressure must not block normal API requests. The buffer is therefore +// bounded and lossy by design. Losing evidence degrades learning; blocking a +// request degrades the service, and the service is the thing that must not +// degrade. +type Emitter struct { + sink Sink + ch chan contract.FluidTelemetry + + dropped atomic.Int64 + written atomic.Int64 + failed atomic.Int64 + stopOnce sync.Once + done chan struct{} + wg sync.WaitGroup +} + +// EmitterOptions configures buffering and delivery. +type EmitterOptions struct { + // Buffer is the number of events held in memory. When it fills, new events + // are dropped rather than queued. + Buffer int + // Workers is the number of concurrent deliveries. + Workers int + // WriteTimeout bounds a single sink write. + WriteTimeout time.Duration +} + +// NewEmitter starts an emitter delivering into sink. +func NewEmitter(sink Sink, opts EmitterOptions) *Emitter { + if opts.Buffer <= 0 { + opts.Buffer = 4096 + } + if opts.Workers <= 0 { + opts.Workers = 2 + } + if opts.WriteTimeout <= 0 { + opts.WriteTimeout = 5 * time.Second + } + + e := &Emitter{ + sink: sink, + ch: make(chan contract.FluidTelemetry, opts.Buffer), + done: make(chan struct{}), + } + + for i := 0; i < opts.Workers; i++ { + e.wg.Add(1) + go e.run(opts.WriteTimeout) + } + return e +} + +// Emit queues an event. It never blocks and never returns an error: a caller on +// the request path has no useful response to a telemetry failure, and giving it +// one invites handling that blocks. +func (e *Emitter) Emit(ev contract.FluidTelemetry) { + select { + case e.ch <- ev: + default: + e.dropped.Add(1) + } +} + +func (e *Emitter) run(timeout time.Duration) { + defer e.wg.Done() + for { + select { + case ev, ok := <-e.ch: + if !ok { + return + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + if err := e.sink.Write(ctx, ev); err != nil { + e.failed.Add(1) + } else { + e.written.Add(1) + } + cancel() + case <-e.done: + // Drain what is already buffered, then stop. + for { + select { + case ev := <-e.ch: + ctx, cancel := context.WithTimeout(context.Background(), timeout) + if err := e.sink.Write(ctx, ev); err != nil { + e.failed.Add(1) + } else { + e.written.Add(1) + } + cancel() + default: + return + } + } + } + } +} + +// Close stops delivery after draining the buffer. +func (e *Emitter) Close() { + e.stopOnce.Do(func() { close(e.done) }) + e.wg.Wait() +} + +// Stats reports emitter health. +// +// FLUID must observe itself (ArchitectureBlueprint.md section 40), and a +// silently lossy telemetry path would make every downstream pressure count +// quietly wrong. Dropped events are a first-class operational metric. +type Stats struct { + Written int64 + Dropped int64 + Failed int64 +} + +// Stats returns a snapshot of delivery counters. +func (e *Emitter) Stats() Stats { + return Stats{ + Written: e.written.Load(), + Dropped: e.dropped.Load(), + Failed: e.failed.Load(), + } +} diff --git a/internal/runtime/telemetry_test.go b/internal/runtime/telemetry_test.go new file mode 100644 index 0000000..e82687d --- /dev/null +++ b/internal/runtime/telemetry_test.go @@ -0,0 +1,101 @@ +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) + } +} diff --git a/workplans/FLUID-WP-0003-deterministic-data-plane.md b/workplans/FLUID-WP-0003-deterministic-data-plane.md index 496a4d1..ddc1002 100644 --- a/workplans/FLUID-WP-0003-deterministic-data-plane.md +++ b/workplans/FLUID-WP-0003-deterministic-data-plane.md @@ -26,7 +26,7 @@ above it is dead. ```task id: FLUID-WP-0003-T01 -status: todo +status: done priority: high state_hub_task_id: "0dac6ab6-f3e4-5234-bb3c-2393558d8590" ``` @@ -38,7 +38,7 @@ shape limits. Blueprint §5.1 — the gateway must not invent interface semantic ```task id: FLUID-WP-0003-T02 -status: todo +status: done priority: high state_hub_task_id: "6d8333ac-1bad-59a4-943d-b83819c8127a" ``` @@ -51,7 +51,7 @@ without an auditable reason is a defect. ```task id: FLUID-WP-0003-T03 -status: todo +status: done priority: high state_hub_task_id: "1f042c4f-9e3f-50fa-985c-e4f2ebb3b38a" ``` @@ -76,7 +76,7 @@ of the revision artifact and is content-addressed. ```task id: FLUID-WP-0003-T05 -status: todo +status: done priority: high state_hub_task_id: "0d62a681-3154-50af-894a-98b35db7b040" ``` @@ -89,7 +89,7 @@ language-agnostic promise is kept. ```task id: FLUID-WP-0003-T06 -status: todo +status: done priority: medium state_hub_task_id: "cc3c2a03-b24c-51de-b695-8aa9bfae85ab" ``` @@ -101,7 +101,7 @@ detail (§5.7). ```task id: FLUID-WP-0003-T07 -status: todo +status: done priority: high state_hub_task_id: "e10a5011-cb7e-5f1c-9a0d-0cfd2d535d13" ```