diff --git a/go.mod b/go.mod index 87015ca..c33a0cd 100644 --- a/go.mod +++ b/go.mod @@ -1,21 +1,3 @@ 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 deleted file mode 100644 index 7b76d67..0000000 --- a/go.sum +++ /dev/null @@ -1,32 +0,0 @@ -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 deleted file mode 100644 index 33256eb..0000000 --- a/internal/runtime/connector.go +++ /dev/null @@ -1,248 +0,0 @@ -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 deleted file mode 100644 index 2ed1089..0000000 --- a/internal/runtime/gateway.go +++ /dev/null @@ -1,284 +0,0 @@ -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 deleted file mode 100644 index 7042310..0000000 --- a/internal/runtime/gateway_test.go +++ /dev/null @@ -1,215 +0,0 @@ -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 deleted file mode 100644 index 0a3d6db..0000000 --- a/internal/runtime/registry.go +++ /dev/null @@ -1,170 +0,0 @@ -// 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 deleted file mode 100644 index 03f0aed..0000000 --- a/internal/runtime/resolver.go +++ /dev/null @@ -1,212 +0,0 @@ -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 deleted file mode 100644 index 20c7ca2..0000000 --- a/internal/runtime/resolver_test.go +++ /dev/null @@ -1,275 +0,0 @@ -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 deleted file mode 100644 index b0ff49c..0000000 --- a/internal/runtime/response.go +++ /dev/null @@ -1,91 +0,0 @@ -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 deleted file mode 100644 index 974de8f..0000000 --- a/internal/runtime/telemetry.go +++ /dev/null @@ -1,143 +0,0 @@ -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 deleted file mode 100644 index e82687d..0000000 --- a/internal/runtime/telemetry_test.go +++ /dev/null @@ -1,101 +0,0 @@ -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 ddc1002..496a4d1 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: done +status: todo 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: done +status: todo 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: done +status: todo 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: done +status: todo 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: done +status: todo 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: done +status: todo priority: high state_hub_task_id: "e10a5011-cb7e-5f1c-9a0d-0cfd2d535d13" ```