diff --git a/go.mod b/go.mod index c33a0cd..87015ca 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,21 @@ module github.com/tegwick/fluid-core go 1.22.2 + +require ( + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/sys v0.19.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect + modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect + modernc.org/libc v1.49.3 // indirect + modernc.org/mathutil v1.6.0 // indirect + modernc.org/memory v1.8.0 // indirect + modernc.org/sqlite v1.29.10 // indirect + modernc.org/strutil v1.2.0 // indirect + modernc.org/token v1.1.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..7b76d67 --- /dev/null +++ b/go.sum @@ -0,0 +1,32 @@ +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= +github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o= +golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 h1:5D53IMaUuA5InSeMu9eJtlQXS2NxAhyWQvkKEgXZhHI= +modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6/go.mod h1:Qz0X07sNOR1jWYCrJMEnbW/X55x206Q7Vt4mz6/wHp4= +modernc.org/libc v1.49.3 h1:j2MRCRdwJI2ls/sGbeSk0t2bypOG/uvPZUsGQFDulqg= +modernc.org/libc v1.49.3/go.mod h1:yMZuGkn7pXbKfoT/M35gFJOAEdSKdxL0q64sF7KqCDo= +modernc.org/mathutil v1.6.0 h1:fRe9+AmYlaej+64JsEEhoWuAYBkOtQiMEU7n/XgfYi4= +modernc.org/mathutil v1.6.0/go.mod h1:Ui5Q9q1TR2gFm0AQRqQUaBWFLAhQpCwNcuhBOSedWPo= +modernc.org/memory v1.8.0 h1:IqGTL6eFMaDZZhEWwcREgeMXYwmW83LYW8cROZYkg+E= +modernc.org/memory v1.8.0/go.mod h1:XPZ936zp5OMKGWPqbD3JShgd/ZoQ7899TUuQqxY+peU= +modernc.org/sqlite v1.29.10 h1:3u93dz83myFnMilBGCOLbr+HjklS6+5rJLx4q86RDAg= +modernc.org/sqlite v1.29.10/go.mod h1:ItX2a1OVGgNsFh6Dv60JQvGfJfTPHPVpV6DF59akYOA= +modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA= +modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= 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/registry.go b/internal/runtime/registry.go new file mode 100644 index 0000000..0a3d6db --- /dev/null +++ b/internal/runtime/registry.go @@ -0,0 +1,170 @@ +// Package runtime implements the FLUID deterministic data plane: the gateway, +// revision resolution, routing, backend connection and telemetry emission. +// +// Nothing here may depend on the evolution control plane to serve a request. +// ArchitectureBlueprint.md section 2 makes this an invariant: the interface +// runtime must continue to function when the Daimon, the model provider, the +// hypothesis store, the experiment controller and the AI budget are all +// unavailable. +package runtime + +import ( + "errors" + "fmt" + "sync" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Registry is the gateway's cached view of published control-plane state. +// +// It is deliberately a snapshot rather than a client. When the control plane +// dies the registry keeps answering from what it last held, which is what +// ArchitectureBlueprint.md section 34.6 requires: the runtime continues using +// cached published configuration, and no new promotions occur until +// control-plane consistency is restored. +type Registry struct { + mu sync.RWMutex + iface contract.InterfaceID + revisions map[contract.RevisionID]contract.Revision + policy contract.RoutingPolicy + hasPolicy bool +} + +// NewRegistry returns an empty registry for one interface. +func NewRegistry(iface contract.InterfaceID) *Registry { + return &Registry{ + iface: iface, + revisions: make(map[contract.RevisionID]contract.Revision), + } +} + +var ( + // ErrUnknownRevision is returned for a revision the registry has never seen. + ErrUnknownRevision = errors.New("unknown revision") + // ErrRevisionNotRoutable is returned for a revision that exists but must not + // receive traffic. + ErrRevisionNotRoutable = errors.New("revision not routable") + // ErrNoPolicy is returned before any routing policy has been loaded. + ErrNoPolicy = errors.New("no routing policy loaded") + // ErrWrongInterface guards against loading another interface's artifacts. + ErrWrongInterface = errors.New("artifact belongs to a different interface") + // ErrStalePolicy is returned when an older policy generation is offered. + ErrStalePolicy = errors.New("routing policy generation is not newer") +) + +// PutRevision publishes a revision descriptor into the registry. +// +// The descriptor is expected to have been signature-verified already; this +// method enforces only the structural conditions the router depends on. +func (r *Registry) PutRevision(d contract.Revision) error { + if d.Interface != r.iface { + return fmt.Errorf("%w: descriptor is for %q, registry serves %q", + ErrWrongInterface, d.Interface, r.iface) + } + if d.Runtime.Upstream == "" { + return fmt.Errorf("revision %s: descriptor has no runtime upstream", d.ID) + } + if !d.State.Valid() { + return fmt.Errorf("revision %s: unknown state %q", d.ID, d.State) + } + + r.mu.Lock() + defer r.mu.Unlock() + r.revisions[d.ID] = d + return nil +} + +// Revision returns a published descriptor. +func (r *Registry) Revision(id contract.RevisionID) (contract.Revision, error) { + r.mu.RLock() + defer r.mu.RUnlock() + d, ok := r.revisions[id] + if !ok { + return contract.Revision{}, fmt.Errorf("%w: %s", ErrUnknownRevision, id) + } + return d, nil +} + +// PutPolicy installs a routing policy. +// +// Generations are monotonic: an older policy is refused rather than applied. +// Without this a delayed delivery could silently roll traffic back to a +// superseded allocation, which would corrupt an in-flight experiment's +// measurement window. +func (r *Registry) PutPolicy(p contract.RoutingPolicy) error { + if p.Interface != r.iface { + return fmt.Errorf("%w: policy is for %q, registry serves %q", + ErrWrongInterface, p.Interface, r.iface) + } + + r.mu.Lock() + defer r.mu.Unlock() + + if r.hasPolicy && p.Generation <= r.policy.Generation { + return fmt.Errorf("%w: offered %d, holding %d", + ErrStalePolicy, p.Generation, r.policy.Generation) + } + r.policy = p + r.hasPolicy = true + return nil +} + +// Policy returns the current routing policy. +func (r *Registry) Policy() (contract.RoutingPolicy, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if !r.hasPolicy { + return contract.RoutingPolicy{}, ErrNoPolicy + } + return r.policy, nil +} + +// routableStates lists the descriptor states the router may send traffic to. +// +// ArchitectureBlueprint.md section 5.3 requires rejecting unpublished, failed +// and retired revisions. "created" is unpublished; "verified" has passed tests +// but has not been exposed; "retired" is finished. +var routableStates = map[contract.RevisionState]bool{ + contract.RevisionStateExperiment: true, + contract.RevisionStateCandidate: true, + contract.RevisionStateStable: true, + contract.RevisionStateDeprecated: true, +} + +// CheckRoutable reports whether a revision may currently receive traffic from +// the given cohort. +func (r *Registry) CheckRoutable(id contract.RevisionID, cohort contract.CohortID) error { + d, err := r.Revision(id) + if err != nil { + return err + } + + if !routableStates[d.State] { + return fmt.Errorf("%w: %s is %s", ErrRevisionNotRoutable, id, d.State) + } + if d.Policy.SecurityCheck != contract.RevisionPolicySecurityCheckPassed { + return fmt.Errorf("%w: %s has security_check=%s", + ErrRevisionNotRoutable, id, d.Policy.SecurityCheck) + } + if d.Policy.PolicyCheck != nil && *d.Policy.PolicyCheck == contract.RevisionPolicyPolicyCheckFailed { + return fmt.Errorf("%w: %s failed its policy check", ErrRevisionNotRoutable, id) + } + + if d.Routing != nil && len(d.Routing.EligibleCohorts) > 0 { + if !containsCohort(d.Routing.EligibleCohorts, cohort) { + return fmt.Errorf("%w: cohort %q is not eligible for %s", + ErrRevisionNotRoutable, cohort, id) + } + } + return nil +} + +func containsCohort(list []contract.CohortID, want contract.CohortID) bool { + for _, c := range list { + if c == want { + return true + } + } + return false +} diff --git a/internal/runtime/resolver.go b/internal/runtime/resolver.go new file mode 100644 index 0000000..03f0aed --- /dev/null +++ b/internal/runtime/resolver.go @@ -0,0 +1,212 @@ +package runtime + +import ( + "fmt" + "hash/fnv" + "sort" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Request is the subset of an inbound request that revision resolution may +// consider. Nothing else is allowed to influence the decision: resolution must +// be a pure function of these fields and the loaded policy, or it stops being +// auditable (ArchitectureBlueprint.md section 5.2). +type Request struct { + // ExplicitRevision is a revision the consumer asked for by name. + ExplicitRevision contract.RevisionID + // BoundRevision comes from a client contract binding. + BoundRevision contract.RevisionID + // Cohort is the consumer's cohort assignment. + Cohort contract.CohortID + // Tenant identifies the calling tenant, where the interface is multi-tenant. + Tenant string + // ConsumerRef is the pseudonymous, stable consumer identity used to keep a + // long-lived consumer on one side of an experiment. + ConsumerRef string + // CorrelationID ties this request to its telemetry. + CorrelationID string +} + +// Resolution is the outcome of revision resolution, including why. +// +// The reason is not decoration. Blueprint section 5.2 requires resolution to be +// auditable, and "which revision served this request" is unanswerable later +// without recording how it was chosen. +type Resolution struct { + Revision contract.RevisionID + Reason contract.FluidTelemetryResolutionReason + Experiment *contract.ExperimentID + // PolicyGeneration records which policy produced this decision. + PolicyGeneration int64 +} + +// Resolver implements the deterministic precedence chain. +type Resolver struct { + registry *Registry + // allowExplicit controls whether consumers may pin a revision by name. Some + // interfaces want this for migration testing; others must not expose it. + allowExplicit bool +} + +// NewResolver returns a resolver over reg. +func NewResolver(reg *Registry, allowExplicit bool) *Resolver { + return &Resolver{registry: reg, allowExplicit: allowExplicit} +} + +// Resolve selects the revision that will serve req. +// +// The order is fixed by ArchitectureBlueprint.md section 5.2: +// +// explicit revision -> bound client contract -> experiment assignment -> stable default +// +// Each step is skipped rather than failed when the candidate is not routable, +// so a retired pin or an ineligible cohort degrades to the default instead of +// erroring the request. +func (r *Resolver) Resolve(req Request) (Resolution, error) { + policy, err := r.registry.Policy() + if err != nil { + return Resolution{}, err + } + + if r.allowExplicit && req.ExplicitRevision != "" { + if err := r.registry.CheckRoutable(req.ExplicitRevision, req.Cohort); err != nil { + // An explicit request for something unroutable is a consumer error + // worth surfacing, not something to silently reinterpret. + return Resolution{}, fmt.Errorf("explicit revision %s: %w", req.ExplicitRevision, err) + } + return Resolution{ + Revision: req.ExplicitRevision, + Reason: contract.FluidTelemetryResolutionReasonExplicitRevision, + PolicyGeneration: policy.Generation, + }, nil + } + + if req.BoundRevision != "" { + if err := r.registry.CheckRoutable(req.BoundRevision, req.Cohort); err == nil { + return Resolution{ + Revision: req.BoundRevision, + Reason: contract.FluidTelemetryResolutionReasonBoundContract, + PolicyGeneration: policy.Generation, + }, nil + } + } + + if rule, ok := matchRule(policy.Rules, req); ok { + chosen, ok := allocate(rule, req, policy.DefaultRevision) + if ok { + if err := r.registry.CheckRoutable(chosen, req.Cohort); err == nil { + reason := contract.FluidTelemetryResolutionReasonCohortRule + if rule.Experiment != nil { + reason = contract.FluidTelemetryResolutionReasonExperimentAssignment + } + return Resolution{ + Revision: chosen, + Reason: reason, + Experiment: rule.Experiment, + PolicyGeneration: policy.Generation, + }, nil + } + } + } + + if err := r.registry.CheckRoutable(policy.DefaultRevision, req.Cohort); err != nil { + return Resolution{}, fmt.Errorf("default revision %s: %w", policy.DefaultRevision, err) + } + return Resolution{ + Revision: policy.DefaultRevision, + Reason: contract.FluidTelemetryResolutionReasonStableDefault, + PolicyGeneration: policy.Generation, + }, nil +} + +// matchRule returns the first rule matching the request. Rules are evaluated in +// document order and the first match wins, so policy authors control precedence +// by ordering rather than by scoring. +func matchRule(rules []contract.RoutingPolicyRulesItem, req Request) (contract.RoutingPolicyRulesItem, bool) { + for _, rule := range rules { + if rule.Cohort != nil && *rule.Cohort != req.Cohort { + continue + } + if rule.Tenant != "" && rule.Tenant != req.Tenant { + continue + } + return rule, true + } + return contract.RoutingPolicyRulesItem{}, false +} + +// allocate picks a revision from a rule's traffic shares. +// +// Assignment is a deterministic function of the sticky key, so a given consumer +// lands on the same side of an experiment for its whole duration. Random +// per-request assignment would make within-consumer comparisons meaningless and +// would let a client observe both revisions at once. +func allocate(rule contract.RoutingPolicyRulesItem, req Request, fallback contract.RevisionID) (contract.RevisionID, bool) { + if len(rule.Allocation) == 0 { + return "", false + } + + // Sorting makes the traversal order independent of Go's map iteration, which + // is what turns a hash bucket into a stable assignment. + ids := make([]string, 0, len(rule.Allocation)) + var total float64 + for id, share := range rule.Allocation { + ids = append(ids, id) + total += float64(share) + } + sort.Strings(ids) + + if total <= 0 { + return "", false + } + + key := stickyKey(rule, req) + position := bucket(key) * total + + var cumulative float64 + for _, id := range ids { + cumulative += float64(rule.Allocation[id]) + if position < cumulative { + return contract.RevisionID(id), true + } + } + // Floating-point drift at the top of the range. + return contract.RevisionID(ids[len(ids)-1]), true +} + +// stickyKey chooses what keeps a consumer on one side of an experiment. +func stickyKey(rule contract.RoutingPolicyRulesItem, req Request) string { + mode := contract.RoutingPolicyRulesItemStickyByConsumerID + if rule.StickyBy != nil { + mode = *rule.StickyBy + } + + var subject string + switch mode { + case contract.RoutingPolicyRulesItemStickyByTenant: + subject = req.Tenant + case contract.RoutingPolicyRulesItemStickyByCorrelationID: + subject = req.CorrelationID + case contract.RoutingPolicyRulesItemStickyByNone: + subject = req.CorrelationID + default: + subject = req.ConsumerRef + } + + // Namespacing by experiment stops one consumer from landing in the same + // arm of every concurrent experiment, which would confound their results. + if rule.Experiment != nil { + return string(*rule.Experiment) + "\x00" + subject + } + return subject +} + +// bucket maps a key into [0, 1). +func bucket(key string) float64 { + h := fnv.New64a() + _, _ = h.Write([]byte(key)) + // 53 bits keeps the result exactly representable as a float64. + const mask = 1<<53 - 1 + return float64(h.Sum64()&mask) / float64(mask+1) +} diff --git a/internal/runtime/resolver_test.go b/internal/runtime/resolver_test.go new file mode 100644 index 0000000..20c7ca2 --- /dev/null +++ b/internal/runtime/resolver_test.go @@ -0,0 +1,275 @@ +package runtime + +import ( + "errors" + "testing" + + "github.com/tegwick/fluid-core/internal/contract" +) + +const testInterface contract.InterfaceID = "hall-publishing" + +// descriptor builds a minimal routable revision descriptor. +func descriptor(id contract.RevisionID, state contract.RevisionState, cohorts ...contract.CohortID) contract.Revision { + d := contract.Revision{ + SchemaVersion: "0.1", + ID: id, + Interface: testInterface, + State: state, + Contract: contract.RevisionContract{ + Type: contract.RevisionContractTypeOpenapi, + Digest: contract.Digest("sha256:" + zeros(64)), + }, + Runtime: contract.RevisionRuntime{Upstream: "http://adapter:8080"}, + Intent: contract.RevisionIntent{Version: "IEI-1"}, + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityAdditive, + SecurityCheck: contract.RevisionPolicySecurityCheckPassed, + }, + } + if len(cohorts) > 0 { + d.Routing = &contract.RevisionRouting{EligibleCohorts: cohorts} + } + return d +} + +func zeros(n int) string { + b := make([]byte, n) + for i := range b { + b[i] = '0' + } + return string(b) +} + +func policyWith(defaultRev contract.RevisionID, gen int64, rules ...contract.RoutingPolicyRulesItem) contract.RoutingPolicy { + return contract.RoutingPolicy{ + SchemaVersion: "0.1", + Interface: testInterface, + Generation: gen, + DefaultRevision: defaultRev, + Rules: rules, + } +} + +func newFixture(t *testing.T, revs ...contract.Revision) *Registry { + t.Helper() + reg := NewRegistry(testInterface) + for _, r := range revs { + if err := reg.PutRevision(r); err != nil { + t.Fatalf("PutRevision(%s): %v", r.ID, err) + } + } + return reg +} + +func TestResolvePrecedence(t *testing.T) { + reg := newFixture(t, + descriptor("R-1", contract.RevisionStateStable), + descriptor("R-2", contract.RevisionStateExperiment), + descriptor("R-3", contract.RevisionStateCandidate), + ) + if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil { + t.Fatal(err) + } + + res := NewResolver(reg, true) + + t.Run("explicit wins", func(t *testing.T) { + got, err := res.Resolve(Request{ExplicitRevision: "R-3", BoundRevision: "R-2"}) + if err != nil { + t.Fatal(err) + } + if got.Revision != "R-3" { + t.Errorf("got %s, want R-3", got.Revision) + } + if got.Reason != contract.FluidTelemetryResolutionReasonExplicitRevision { + t.Errorf("reason = %s", got.Reason) + } + }) + + t.Run("bound contract beats default", func(t *testing.T) { + got, err := res.Resolve(Request{BoundRevision: "R-2"}) + if err != nil { + t.Fatal(err) + } + if got.Revision != "R-2" || got.Reason != contract.FluidTelemetryResolutionReasonBoundContract { + t.Errorf("got %s via %s, want R-2 via bound_contract", got.Revision, got.Reason) + } + }) + + t.Run("falls through to stable default", func(t *testing.T) { + got, err := res.Resolve(Request{}) + if err != nil { + t.Fatal(err) + } + if got.Revision != "R-1" || got.Reason != contract.FluidTelemetryResolutionReasonStableDefault { + t.Errorf("got %s via %s, want R-1 via stable_default", got.Revision, got.Reason) + } + }) + + t.Run("explicit disabled is ignored", func(t *testing.T) { + strict := NewResolver(reg, false) + got, err := strict.Resolve(Request{ExplicitRevision: "R-3"}) + if err != nil { + t.Fatal(err) + } + if got.Revision != "R-1" { + t.Errorf("got %s, want the default R-1 when pinning is disabled", got.Revision) + } + }) +} + +func TestUnroutableRevisionsAreRefused(t *testing.T) { + reg := newFixture(t, + descriptor("R-1", contract.RevisionStateStable), + descriptor("R-created", contract.RevisionStateCreated), + descriptor("R-retired", contract.RevisionStateRetired), + ) + failed := descriptor("R-insecure", contract.RevisionStateStable) + failed.Policy.SecurityCheck = contract.RevisionPolicySecurityCheckFailed + if err := reg.PutRevision(failed); err != nil { + t.Fatal(err) + } + if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil { + t.Fatal(err) + } + res := NewResolver(reg, true) + + for _, id := range []contract.RevisionID{"R-created", "R-retired", "R-insecure"} { + if _, err := res.Resolve(Request{ExplicitRevision: id}); err == nil { + t.Errorf("resolving %s should have been refused", id) + } else if !errors.Is(err, ErrRevisionNotRoutable) { + t.Errorf("resolving %s: got %v, want ErrRevisionNotRoutable", id, err) + } + } +} + +func TestBoundRevisionDegradesToDefault(t *testing.T) { + // A consumer pinned to a revision that has since retired should keep being + // served rather than start failing. + reg := newFixture(t, + descriptor("R-1", contract.RevisionStateStable), + descriptor("R-old", contract.RevisionStateRetired), + ) + if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil { + t.Fatal(err) + } + + got, err := NewResolver(reg, false).Resolve(Request{BoundRevision: "R-old"}) + if err != nil { + t.Fatalf("a retired binding should degrade, not fail: %v", err) + } + if got.Revision != "R-1" { + t.Errorf("got %s, want R-1", got.Revision) + } +} + +func TestCohortEligibility(t *testing.T) { + reg := newFixture(t, + descriptor("R-1", contract.RevisionStateStable), + descriptor("R-2", contract.RevisionStateExperiment, "coding-agents"), + ) + if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil { + t.Fatal(err) + } + + if err := reg.CheckRoutable("R-2", "coding-agents"); err != nil { + t.Errorf("eligible cohort refused: %v", err) + } + if err := reg.CheckRoutable("R-2", "partner-integrations"); err == nil { + t.Error("ineligible cohort accepted") + } +} + +func TestAllocationIsStickyAndProportional(t *testing.T) { + reg := newFixture(t, + descriptor("R-1", contract.RevisionStateStable), + descriptor("R-2", contract.RevisionStateExperiment), + ) + exp := contract.ExperimentID("E-1") + rule := contract.RoutingPolicyRulesItem{ + Experiment: &exp, + Allocation: map[string]contract.UnitInterval{"R-1": 0.9, "R-2": 0.1}, + } + if err := reg.PutPolicy(policyWith("R-1", 1, rule)); err != nil { + t.Fatal(err) + } + res := NewResolver(reg, false) + + // Stickiness: the same consumer must resolve identically every time. + first, err := res.Resolve(Request{ConsumerRef: "consumer-42"}) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 50; i++ { + again, err := res.Resolve(Request{ConsumerRef: "consumer-42"}) + if err != nil { + t.Fatal(err) + } + if again.Revision != first.Revision { + t.Fatalf("assignment drifted: %s then %s", first.Revision, again.Revision) + } + } + + // Proportionality: roughly a tenth of consumers should see the candidate. + const n = 4000 + candidate := 0 + for i := 0; i < n; i++ { + got, err := res.Resolve(Request{ConsumerRef: consumerName(i)}) + if err != nil { + t.Fatal(err) + } + if got.Revision == "R-2" { + candidate++ + } + if got.Experiment == nil || *got.Experiment != exp { + t.Fatalf("experiment not recorded on resolution for consumer %d", i) + } + } + share := float64(candidate) / n + if share < 0.07 || share > 0.13 { + t.Errorf("candidate share %.3f, want approximately 0.10", share) + } +} + +func consumerName(i int) string { + digits := "0123456789" + out := []byte("consumer-") + if i == 0 { + return string(append(out, '0')) + } + var rev []byte + for i > 0 { + rev = append(rev, digits[i%10]) + i /= 10 + } + for j := len(rev) - 1; j >= 0; j-- { + out = append(out, rev[j]) + } + return string(out) +} + +func TestStalePolicyRefused(t *testing.T) { + reg := newFixture(t, descriptor("R-1", contract.RevisionStateStable)) + if err := reg.PutPolicy(policyWith("R-1", 5)); err != nil { + t.Fatal(err) + } + if err := reg.PutPolicy(policyWith("R-1", 4)); !errors.Is(err, ErrStalePolicy) { + t.Errorf("older generation accepted: %v", err) + } + if err := reg.PutPolicy(policyWith("R-1", 5)); !errors.Is(err, ErrStalePolicy) { + t.Errorf("equal generation accepted: %v", err) + } + if err := reg.PutPolicy(policyWith("R-1", 6)); err != nil { + t.Errorf("newer generation refused: %v", err) + } +} + +func TestWrongInterfaceRefused(t *testing.T) { + reg := NewRegistry(testInterface) + other := descriptor("R-1", contract.RevisionStateStable) + other.Interface = "some-other-api" + if err := reg.PutRevision(other); !errors.Is(err, ErrWrongInterface) { + t.Errorf("foreign descriptor accepted: %v", err) + } +} 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" ```