diff --git a/.sqlx/query-2cb997fda736985f573a6b111a7c4d8a83721fcffbea9413dee28811f4a2331f.json b/.sqlx/query-2cb997fda736985f573a6b111a7c4d8a83721fcffbea9413dee28811f4a2331f.json new file mode 100644 index 0000000..d32e40d --- /dev/null +++ b/.sqlx/query-2cb997fda736985f573a6b111a7c4d8a83721fcffbea9413dee28811f4a2331f.json @@ -0,0 +1,208 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT e.id, e.environment_id, e.flag_id, f.key AS flag_key, f.variants,\n e.key, e.name, e.description, e.metric_key, e.control_variant,\n e.state, e.started_at, e.stopped_at, e.version, e.created_at, e.updated_at\n FROM experiments e\n JOIN flags f ON f.id = e.flag_id\n WHERE e.environment_id = $1 ORDER BY e.key", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid", + "origin": { + "Table": { + "table": "experiments", + "name": "id" + } + } + }, + { + "ordinal": 1, + "name": "environment_id", + "type_info": "Uuid", + "origin": { + "Table": { + "table": "experiments", + "name": "environment_id" + } + } + }, + { + "ordinal": 2, + "name": "flag_id", + "type_info": "Uuid", + "origin": { + "Table": { + "table": "experiments", + "name": "flag_id" + } + } + }, + { + "ordinal": 3, + "name": "flag_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "flags", + "name": "key" + } + } + }, + { + "ordinal": 4, + "name": "variants", + "type_info": "Jsonb", + "origin": { + "Table": { + "table": "flags", + "name": "variants" + } + } + }, + { + "ordinal": 5, + "name": "key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "key" + } + } + }, + { + "ordinal": 6, + "name": "name", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "name" + } + } + }, + { + "ordinal": 7, + "name": "description", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "description" + } + } + }, + { + "ordinal": 8, + "name": "metric_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "metric_key" + } + } + }, + { + "ordinal": 9, + "name": "control_variant", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "control_variant" + } + } + }, + { + "ordinal": 10, + "name": "state", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "state" + } + } + }, + { + "ordinal": 11, + "name": "started_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "started_at" + } + } + }, + { + "ordinal": 12, + "name": "stopped_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "stopped_at" + } + } + }, + { + "ordinal": 13, + "name": "version", + "type_info": "Int8", + "origin": { + "Table": { + "table": "experiments", + "name": "version" + } + } + }, + { + "ordinal": 14, + "name": "created_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "created_at" + } + } + }, + { + "ordinal": 15, + "name": "updated_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "updated_at" + } + } + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + false, + true, + false, + false, + false, + true, + true, + false, + false, + false + ] + }, + "hash": "2cb997fda736985f573a6b111a7c4d8a83721fcffbea9413dee28811f4a2331f" +} diff --git a/.sqlx/query-3908cb7a1dcd376eb4283f3699447444a1c4a24b5fa85f40caca1d298564ca4a.json b/.sqlx/query-3908cb7a1dcd376eb4283f3699447444a1c4a24b5fa85f40caca1d298564ca4a.json new file mode 100644 index 0000000..3572760 --- /dev/null +++ b/.sqlx/query-3908cb7a1dcd376eb4283f3699447444a1c4a24b5fa85f40caca1d298564ca4a.json @@ -0,0 +1,209 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT e.id, e.environment_id, e.flag_id, f.key AS flag_key, f.variants,\n e.key, e.name, e.description, e.metric_key, e.control_variant,\n e.state, e.started_at, e.stopped_at, e.version, e.created_at, e.updated_at\n FROM experiments e\n JOIN flags f ON f.id = e.flag_id\n WHERE e.environment_id = $1 AND e.key = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid", + "origin": { + "Table": { + "table": "experiments", + "name": "id" + } + } + }, + { + "ordinal": 1, + "name": "environment_id", + "type_info": "Uuid", + "origin": { + "Table": { + "table": "experiments", + "name": "environment_id" + } + } + }, + { + "ordinal": 2, + "name": "flag_id", + "type_info": "Uuid", + "origin": { + "Table": { + "table": "experiments", + "name": "flag_id" + } + } + }, + { + "ordinal": 3, + "name": "flag_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "flags", + "name": "key" + } + } + }, + { + "ordinal": 4, + "name": "variants", + "type_info": "Jsonb", + "origin": { + "Table": { + "table": "flags", + "name": "variants" + } + } + }, + { + "ordinal": 5, + "name": "key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "key" + } + } + }, + { + "ordinal": 6, + "name": "name", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "name" + } + } + }, + { + "ordinal": 7, + "name": "description", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "description" + } + } + }, + { + "ordinal": 8, + "name": "metric_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "metric_key" + } + } + }, + { + "ordinal": 9, + "name": "control_variant", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "control_variant" + } + } + }, + { + "ordinal": 10, + "name": "state", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "state" + } + } + }, + { + "ordinal": 11, + "name": "started_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "started_at" + } + } + }, + { + "ordinal": 12, + "name": "stopped_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "stopped_at" + } + } + }, + { + "ordinal": 13, + "name": "version", + "type_info": "Int8", + "origin": { + "Table": { + "table": "experiments", + "name": "version" + } + } + }, + { + "ordinal": 14, + "name": "created_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "created_at" + } + } + }, + { + "ordinal": 15, + "name": "updated_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "experiments", + "name": "updated_at" + } + } + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + false, + true, + false, + false, + false, + true, + true, + false, + false, + false + ] + }, + "hash": "3908cb7a1dcd376eb4283f3699447444a1c4a24b5fa85f40caca1d298564ca4a" +} diff --git a/.sqlx/query-5b52511dc9c9ada9929695c740c266d5b8a83aebc0fe9b811fa094a9ca15eaa3.json b/.sqlx/query-5b52511dc9c9ada9929695c740c266d5b8a83aebc0fe9b811fa094a9ca15eaa3.json new file mode 100644 index 0000000..a7db599 --- /dev/null +++ b/.sqlx/query-5b52511dc9c9ada9929695c740c266d5b8a83aebc0fe9b811fa094a9ca15eaa3.json @@ -0,0 +1,40 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT env.key AS environment_key, e.key\n FROM experiments e\n JOIN environments env ON env.id = e.environment_id\n WHERE e.flag_id = $1\n ORDER BY env.key, e.key\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "environment_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "environments", + "name": "key" + } + } + }, + { + "ordinal": 1, + "name": "key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "key" + } + } + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "5b52511dc9c9ada9929695c740c266d5b8a83aebc0fe9b811fa094a9ca15eaa3" +} diff --git a/.sqlx/query-6e36655558ad952fd32f0a01377e796d194b242d29ac80e1830c4e2f63130b15.json b/.sqlx/query-6e36655558ad952fd32f0a01377e796d194b242d29ac80e1830c4e2f63130b15.json new file mode 100644 index 0000000..9928164 --- /dev/null +++ b/.sqlx/query-6e36655558ad952fd32f0a01377e796d194b242d29ac80e1830c4e2f63130b15.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO experiment_counters (experiment_id, variant, kind, hour, count)\n SELECT e.id, d.variant, d.kind, d.hour, d.count\n FROM UNNEST($2::TEXT[], $3::TEXT[], $4::TEXT[], $5::TIMESTAMPTZ[], $6::BIGINT[])\n AS d(experiment_key, variant, kind, hour, count)\n JOIN experiments e\n ON e.environment_id = $1 AND e.key = d.experiment_key AND e.state = 'running'\n ON CONFLICT (experiment_id, variant, kind, hour)\n DO UPDATE SET count = experiment_counters.count + EXCLUDED.count\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "TextArray", + "TextArray", + "TextArray", + "TimestamptzArray", + "Int8Array" + ] + }, + "nullable": [] + }, + "hash": "6e36655558ad952fd32f0a01377e796d194b242d29ac80e1830c4e2f63130b15" +} diff --git a/.sqlx/query-76a0e1755c781035f94285930971fa0118c5d7e59d9940649f58cdf6c45994b4.json b/.sqlx/query-76a0e1755c781035f94285930971fa0118c5d7e59d9940649f58cdf6c45994b4.json new file mode 100644 index 0000000..39f7392 --- /dev/null +++ b/.sqlx/query-76a0e1755c781035f94285930971fa0118c5d7e59d9940649f58cdf6c45994b4.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE experiments\n SET state = $4,\n started_at = CASE WHEN $4 = 'running' THEN now() ELSE started_at END,\n stopped_at = CASE WHEN $4 = 'stopped' THEN now() ELSE stopped_at END\n WHERE environment_id = $1 AND key = $2 AND state = $3\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "76a0e1755c781035f94285930971fa0118c5d7e59d9940649f58cdf6c45994b4" +} diff --git a/.sqlx/query-9e218bf152147e7eae2d95e2142ad5b0ec3f32c1f778a50e64e4aff14c0faee7.json b/.sqlx/query-9e218bf152147e7eae2d95e2142ad5b0ec3f32c1f778a50e64e4aff14c0faee7.json new file mode 100644 index 0000000..c87eb3d --- /dev/null +++ b/.sqlx/query-9e218bf152147e7eae2d95e2142ad5b0ec3f32c1f778a50e64e4aff14c0faee7.json @@ -0,0 +1,76 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT e.key, f.key AS flag_key, e.metric_key, e.control_variant, e.version\n FROM experiments e\n JOIN flags f ON f.id = e.flag_id\n WHERE e.environment_id = $1 AND e.state = 'running'\n ORDER BY e.key\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "key" + } + } + }, + { + "ordinal": 1, + "name": "flag_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "flags", + "name": "key" + } + } + }, + { + "ordinal": 2, + "name": "metric_key", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "metric_key" + } + } + }, + { + "ordinal": 3, + "name": "control_variant", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiments", + "name": "control_variant" + } + } + }, + { + "ordinal": 4, + "name": "version", + "type_info": "Int8", + "origin": { + "Table": { + "table": "experiments", + "name": "version" + } + } + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + false, + false, + false + ] + }, + "hash": "9e218bf152147e7eae2d95e2142ad5b0ec3f32c1f778a50e64e4aff14c0faee7" +} diff --git a/.sqlx/query-c82e5c51e3ce3ea61e56e6cb25765752b330e9b2e0a18027d2a3eff3ff2ec1e2.json b/.sqlx/query-c82e5c51e3ce3ea61e56e6cb25765752b330e9b2e0a18027d2a3eff3ff2ec1e2.json new file mode 100644 index 0000000..2d2d302 --- /dev/null +++ b/.sqlx/query-c82e5c51e3ce3ea61e56e6cb25765752b330e9b2e0a18027d2a3eff3ff2ec1e2.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE experiments SET started_at = $3\n WHERE environment_id = $1 AND key = $2 AND state = 'running'\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "c82e5c51e3ce3ea61e56e6cb25765752b330e9b2e0a18027d2a3eff3ff2ec1e2" +} diff --git a/.sqlx/query-d31c599c72dfff0d28e5acfed7e98eef210809b7685d0fb919ee4e53bf3ff612.json b/.sqlx/query-d31c599c72dfff0d28e5acfed7e98eef210809b7685d0fb919ee4e53bf3ff612.json new file mode 100644 index 0000000..f0da65e --- /dev/null +++ b/.sqlx/query-d31c599c72dfff0d28e5acfed7e98eef210809b7685d0fb919ee4e53bf3ff612.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE experiments\n SET name = COALESCE($3, name),\n description = CASE WHEN $4 THEN $5 ELSE description END\n WHERE environment_id = $1 AND key = $2\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Text", + "Bool", + "Text" + ] + }, + "nullable": [] + }, + "hash": "d31c599c72dfff0d28e5acfed7e98eef210809b7685d0fb919ee4e53bf3ff612" +} diff --git a/.sqlx/query-d49eb59f5bdcfe4f76c25585d851a86677d7658ac26f35c419adcd05aa59dade.json b/.sqlx/query-d49eb59f5bdcfe4f76c25585d851a86677d7658ac26f35c419adcd05aa59dade.json new file mode 100644 index 0000000..5c0b533 --- /dev/null +++ b/.sqlx/query-d49eb59f5bdcfe4f76c25585d851a86677d7658ac26f35c419adcd05aa59dade.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM experiments WHERE environment_id = $1 AND key = $2 AND state <> 'running'", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "d49eb59f5bdcfe4f76c25585d851a86677d7658ac26f35c419adcd05aa59dade" +} diff --git a/.sqlx/query-dca96744ac7bbc0236cdf49f9cbd972f79c0099909b44d62c8c8d2aa426fdd55.json b/.sqlx/query-dca96744ac7bbc0236cdf49f9cbd972f79c0099909b44d62c8c8d2aa426fdd55.json new file mode 100644 index 0000000..495cd60 --- /dev/null +++ b/.sqlx/query-dca96744ac7bbc0236cdf49f9cbd972f79c0099909b44d62c8c8d2aa426fdd55.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT variant, kind, SUM(count)::BIGINT AS \"total!\"\n FROM experiment_counters\n WHERE experiment_id = $1\n GROUP BY variant, kind\n ORDER BY variant\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "variant", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiment_counters", + "name": "variant" + } + } + }, + { + "ordinal": 1, + "name": "kind", + "type_info": "Text", + "origin": { + "Table": { + "table": "experiment_counters", + "name": "kind" + } + } + }, + { + "ordinal": 2, + "name": "total!", + "type_info": "Int8", + "origin": "Expression" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + null + ] + }, + "hash": "dca96744ac7bbc0236cdf49f9cbd972f79c0099909b44d62c8c8d2aa426fdd55" +} diff --git a/.sqlx/query-e2c1f7002e0a5112e3b57b18ced5b31276318a9f3fc1b775db436d9f3addb498.json b/.sqlx/query-e2c1f7002e0a5112e3b57b18ced5b31276318a9f3fc1b775db436d9f3addb498.json new file mode 100644 index 0000000..2672fba --- /dev/null +++ b/.sqlx/query-e2c1f7002e0a5112e3b57b18ced5b31276318a9f3fc1b775db436d9f3addb498.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO experiments\n (environment_id, flag_id, key, name, description, metric_key, control_variant)\n VALUES ($1, $2, $3, $4, $5, $6, $7)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid", + "Text", + "Text", + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "e2c1f7002e0a5112e3b57b18ced5b31276318a9f3fc1b775db436d9f3addb498" +} diff --git a/README.md b/README.md index e5d6ef3..cf9efc1 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,10 @@ FlagForge is the service behind that question: people are in it wherever it is referenced. - **Percentage rollouts.** Deterministic and sticky: `user-42` lands in the same bucket on every node, forever, without any shared state. +- **A/B experiments.** Bind a flag to a conversion metric and its variants + become the arms. SDKs count exposures and conversions locally and flush + hourly totals; the dashboard answers with rates, 95 % confidence intervals + and a two-proportion z-test against the control — a verdict, not a hunch. - **Per-environment configuration.** One flag, on in staging, at 5 % in production. Each environment has its own bucketing salt, so a canary in staging does not preselect the same users in production. Every flag is @@ -272,6 +276,18 @@ WASM. The editor is not approximating what the server would do — it is calling user lands on does depend on the salt, and the salt never leaves the server — so the UI says that rather than pretending otherwise. +### Experiments end in a verdict + +An experiment's results: one meter per arm with a confidence interval, and a significance verdict against the control + +An experiment binds a flag to a conversion metric; the flag's variants become +the arms and the traffic you already serve becomes the sample. Each arm gets a +rate, a 95 % Wilson interval drawn on a shared scale, and a two-proportion +z-test against the control — in words, with the exact numbers printed beside +every bar. The lifecycle is one-way (draft → running → stopped) because a +reopened measurement window would average two populations into an answer about +neither, and a stopped experiment's results stay exactly as they ended. + ### The rest of it | | | @@ -334,6 +350,12 @@ let user = EvaluationContext::new(&user_id).with("plan", plan).with("country", c if flags.is_enabled("checkout.v2", &user, false) { render_new_checkout() } + +// When the thing you care about happens, name it. If a running experiment +// measures this metric, the conversion is attributed to whichever variant +// this user was assigned; if none does, it counts toward nothing — so you +// instrument once and experiments come and go without code changes. +flags.track("order.completed", &user); ``` It fetches the environment's configuration once, keeps it in memory, refreshes @@ -492,6 +514,35 @@ Rollouts can also bucket on an attribute instead of the context key (`bucket_by: "account_id"`), which keeps every user of one account on the same side of a rollout. Half a team seeing a new UI is its own kind of bug. +### Experiment counters are bounded by time, not traffic + +An experiments feature usually arrives with an event pipeline: a queue, a raw +event table, a nightly aggregation job. FlagForge stores none of that. SDKs +collapse events into an in-process counter map — one cell per (experiment, +variant, kind) — and flush the *counts*; the server adds them into hourly +cells with a single `INSERT … ON CONFLICT` round trip. A service evaluating a +flag a million times an hour sends the same few dozen bytes as one evaluating +it twice, and the table grows with hours elapsed times variants, not with +traffic. The price, accepted knowingly: individual events are gone, so there +is nothing to re-analyse later. + +Attribution needs no per-user state either. Because assignment is a pure +function, `track("order.completed", &user)` simply re-evaluates the +experiment's flag for that context — the same deterministic bucketing the +exposure went through — so both sides of the rate land on the same arm. The +corner this cuts is visible rather than hidden: a conversion can be tracked +for a context that never evaluated the flag, and the results view reports the +excess instead of laundering it into the rate. + +The statistics are deliberately conventional: a 95 % Wilson interval per arm — +chosen over the normal approximation because young experiments live exactly +where that one misbehaves — and a pooled two-proportion z-test against the +control, with the normal CDF via Abramowitz & Stegun rather than a statistics +crate an order of magnitude larger than the module it would serve. The +approximation's maximum error (1.5 × 10⁻⁷) sits four decimal places below any +p-value cutoff a reader would act on, and the tests pin it against published +values. + ### An invalid flag can never reach the database `flagforge-core::validate` rejects anything the engine could not evaluate @@ -580,26 +631,30 @@ async fn internal_errors_never_leak_their_cause() { cargo test --workspace # needs DATABASE_URL for the integration suite ``` -**219 tests**, in four layers — plus a CI job that runs the quick start above +**244 tests**, in four layers — plus a CI job that runs the quick start above and checks what it promises, because it was once broken while every other check stayed green: -- **Domain (74).** Pure unit tests plus `proptest` properties: buckets stay in +- **Domain (86).** Pure unit tests plus `proptest` properties: buckets stay in range, bucketing is referentially transparent, field boundaries are unambiguous, any full weight partition resolves, and a segment cohort cannot - alias the flag of the same name. -- **HTTP unit (77).** Error mapping, token round trips, tampering detection, + alias the flag of the same name. The experiment statistics live here too — + Wilson intervals that always contain their point estimate, a z-test that is + antisymmetric, and an `erf` checked against published values. +- **HTTP unit (79).** Error mapping, token round trips, tampering detection, the rate limiter's refill maths, key generation, usage-write throttling, and OpenAPI generation (including a check that the domain and storage `Flag` and `Segment` types do not collide into one schema — utoipa keys schemas by type name). -- **Integration (58).** `#[sqlx::test]` gives each test its own freshly +- **Integration (64).** `#[sqlx::test]` gives each test its own freshly migrated database, and the suite drives the *real* router — middleware, extractors and all — via `tower::ServiceExt::oneshot`. Includes the ones a public deployment rests on: that the seed is a no-op the second time, and that the published read-only account is refused every write. -- **SDK (11).** Including the one that matters: a real server on a real socket, - and 300 users evaluated both locally and remotely to prove the two agree. +- **SDK (15).** Including the one that matters: a real server on a real socket, + and 300 users evaluated both locally and remotely to prove the two agree — + and, for experiments, that the counters the server ends up holding equal a + tally the test keeps by hand from local assignment. The integration tests assert the things that would actually hurt: @@ -618,6 +673,9 @@ a_percentage_rollout_is_sticky_and_lands_near_its_target a_revoked_sdk_key_stops_working_immediately two_unrelated_companies_may_share_a_name a_flag_defined_before_an_environment_is_still_configured_in_it +recorded_counters_agree_with_local_assignment +an_experiment_pins_its_flag_against_deletion +duplicate_cells_in_one_batch_are_summed_not_a_server_error ``` CI runs `cargo fmt --check`, `clippy -D warnings`, the full suite against a @@ -719,6 +777,7 @@ flagforge/ │ ├── core/ # domain model + evaluation engine (no I/O) │ │ ├── bucket.rs # deterministic hashing, + property tests │ │ ├── engine.rs # evaluate(flag, context, environment) +│ │ ├── experiment.rs # A/B specs + the statistics that judge them │ │ ├── matcher.rs # targeting operators, segment membership │ │ ├── segment.rs # reusable audiences │ │ └── validate.rs # what may never reach the database @@ -726,7 +785,7 @@ flagforge/ │ ├── api/ # axum handlers, auth, cache, OpenAPI │ │ └── tests/ # integration suite over a real Postgres │ ├── web/ # Leptos dashboard -> WASM, own workspace -│ │ ├── src/pages/ # login, projects, flags, segments, keys, audit +│ │ ├── src/pages/ # login, projects, flags, segments, experiments, keys, audit │ │ └── styles/ # handwritten design system, light + dark │ └── sdk/ # the client a service embeds │ └── tests/ # local decisions vs the server's, user by user diff --git a/crates/api/src/error.rs b/crates/api/src/error.rs index 6cd854c..006e233 100644 --- a/crates/api/src/error.rs +++ b/crates/api/src/error.rs @@ -140,6 +140,9 @@ impl From for ApiError { // honouring it would leave a rule pointing at nothing. The message // names what has to be changed first. in_use @ StorageError::InUse { .. } => Self::Conflict(in_use.to_string()), + // A 409 again: the row exists and the request is well formed, but + // its lifecycle state does not admit this operation. + state @ StorageError::WrongState { .. } => Self::Conflict(state.to_string()), StorageError::Invalid { entity } => { Self::BadRequest(format!("{entity} failed a database constraint")) } diff --git a/crates/api/src/openapi.rs b/crates/api/src/openapi.rs index 262a6f8..da98d60 100644 --- a/crates/api/src/openapi.rs +++ b/crates/api/src/openapi.rs @@ -6,7 +6,9 @@ use utoipa::openapi::security::{ApiKey, ApiKeyValue, HttpAuthScheme, HttpBuilder, SecurityScheme}; use utoipa::{Modify, OpenApi}; -use crate::routes::{audit, auth, evaluate, flags, health, keys, projects, segments}; +use crate::routes::{ + audit, auth, evaluate, events, experiments, flags, health, keys, projects, segments, +}; #[derive(OpenApi)] #[openapi( @@ -24,6 +26,7 @@ use crate::routes::{audit, auth, evaluate, flags, health, keys, projects, segmen (name = "environments", description = "Environments within a project"), (name = "flags", description = "Flag definitions and per-environment configuration"), (name = "segments", description = "Reusable audiences that flag rules reference"), + (name = "experiments", description = "A/B experiments: a flag measured against a metric"), (name = "keys", description = "SDK keys"), (name = "audit", description = "Change history"), (name = "evaluate", description = "The SDK-facing evaluation API"), @@ -54,6 +57,14 @@ use crate::routes::{audit, auth, evaluate, flags, health, keys, projects, segmen segments::get, segments::update, segments::delete, + experiments::list, + experiments::create, + experiments::get, + experiments::update, + experiments::start, + experiments::stop, + experiments::delete, + experiments::results, keys::create, keys::list, keys::revoke, @@ -61,6 +72,7 @@ use crate::routes::{audit, auth, evaluate, flags, health, keys, projects, segmen evaluate::evaluate_all, evaluate::evaluate_one, evaluate::snapshot, + events::ingest, ), components(schemas( crate::error::ProblemDetails, @@ -92,6 +104,18 @@ use crate::routes::{audit, auth, evaluate, flags, health, keys, projects, segmen flagforge_storage::flags::ConfiguredFlag, flagforge_storage::models::Segment, crate::routes::segments::SegmentWithUsage, + flagforge_storage::models::Experiment, + flagforge_storage::models::ExperimentState, + flagforge_storage::models::CounterKind, + flagforge_core::ExperimentSpec, + flagforge_core::VariantCounts, + flagforge_core::VariantResult, + flagforge_core::ConfidenceInterval, + flagforge_core::Comparison, + crate::routes::experiments::ExperimentResults, + crate::routes::events::EventsRequest, + crate::routes::events::EventDelta, + crate::routes::events::EventsResponse, flagforge_storage::models::ApiKey, flagforge_storage::models::KeyScope, flagforge_storage::models::AuditEntry, diff --git a/crates/api/src/routes/evaluate.rs b/crates/api/src/routes/evaluate.rs index af8bf26..97d5a93 100644 --- a/crates/api/src/routes/evaluate.rs +++ b/crates/api/src/routes/evaluate.rs @@ -140,6 +140,11 @@ pub struct SnapshotResponse { /// report it as an ordinary fallthrough — plausible answers, silently /// disagreeing with the server. pub segments: flagforge_core::SegmentSet, + /// The experiments currently running. An SDK that sees one records + /// exposures when it evaluates the flag and attributes tracked metric + /// events to variants; an SDK too old to know the field ignores it and + /// keeps evaluating correctly. + pub experiments: Vec, pub version: i64, pub generated_at: chrono::DateTime, } @@ -185,6 +190,7 @@ pub async fn snapshot( salt: snapshot.salt.clone(), flags: snapshot.flags.clone(), segments: snapshot.segments.clone(), + experiments: snapshot.experiments.clone(), version: snapshot.version, generated_at: snapshot.generated_at, })) diff --git a/crates/api/src/routes/events.rs b/crates/api/src/routes/events.rs new file mode 100644 index 0000000..0841d21 --- /dev/null +++ b/crates/api/src/routes/events.rs @@ -0,0 +1,110 @@ +//! The SDK-facing event ingest endpoint. +//! +//! SDKs aggregate exposures and conversions locally and flush them here as +//! counter increments — counts, not events, so a busy process sends the same +//! few dozen bytes a quiet one does. The server adds them into hourly cells +//! in one round trip. + +use axum::Json; +use axum::extract::State; +use chrono::{DateTime, Utc}; +use flagforge_storage::experiments; +use flagforge_storage::models::{CounterDelta, CounterKind}; +use serde::{Deserialize, Serialize}; + +use crate::auth::SdkIdentity; +use crate::error::{ApiError, ApiResult}; +use crate::state::AppState; + +/// Refuses absurd payloads before they become work; an SDK flushing every few +/// seconds accumulates nowhere near this many distinct cells. +const MAX_EVENTS: usize = 1_000; + +#[derive(Debug, Deserialize, utoipa::ToSchema)] +pub struct EventsRequest { + pub events: Vec, +} + +/// One counter increment. +#[derive(Debug, Deserialize, utoipa::ToSchema)] +pub struct EventDelta { + /// Which experiment, by the key the snapshot announced. + pub experiment_key: String, + /// The variant the context was assigned; the SDK knows it because it + /// evaluates locally with the same deterministic bucketing the server has. + pub variant: String, + pub kind: CounterKind, + /// How many identical events this delta stands for. + #[serde(default = "one")] + pub count: u32, + /// When the events happened; defaults to arrival time. SDKs that batch + /// send the flush interval's timestamps so a counter lands in the hour it + /// belongs to, not the hour the flush ran in. + #[serde(default)] + pub at: Option>, +} + +fn one() -> u32 { + 1 +} + +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct EventsResponse { + /// Hourly counter cells written after aggregation. Zero against a + /// non-empty batch means every event named an experiment that is unknown, + /// stopped, or not yet started — routine when a batch straddles a stop, + /// so it is reported rather than erroring. + pub accepted: u64, + pub received: usize, +} + +/// Ingests a batch of counter increments for the key's environment. +#[utoipa::path( + post, path = "/api/v1/events", tag = "evaluate", + security(("sdk_key" = [])), request_body = EventsRequest, + responses( + (status = 202, description = "Counters updated", body = EventsResponse), + (status = 400, description = "Too many events, or a malformed variant"), + (status = 401, description = "Missing, unknown or revoked SDK key"), + (status = 429, description = "Rate limited"), + ) +)] +pub async fn ingest( + State(state): State, + identity: SdkIdentity, + Json(body): Json, +) -> ApiResult<(axum::http::StatusCode, Json)> { + if body.events.len() > MAX_EVENTS { + return Err(ApiError::BadRequest(format!( + "at most {MAX_EVENTS} events per batch; aggregate client-side and flush more often" + ))); + } + + let received = body.events.len(); + let now = Utc::now(); + + let mut deltas = Vec::with_capacity(received); + for event in body.events { + // Variants come from attacker-controllable input and land in a TEXT + // column; the same charset rule keys live by keeps the table clean. + if event.variant.is_empty() || event.variant.len() > 128 { + return Err(ApiError::BadRequest( + "variant must be between 1 and 128 characters".into(), + )); + } + deltas.push(CounterDelta { + experiment_key: event.experiment_key, + variant: event.variant, + kind: event.kind, + at: event.at.unwrap_or(now), + count: event.count, + }); + } + + let accepted = + experiments::ingest_counters(&state.pool, identity.environment_id(), &deltas).await?; + + metrics::counter!("flagforge_experiment_events_total").increment(accepted); + + Ok((axum::http::StatusCode::ACCEPTED, Json(EventsResponse { accepted, received }))) +} diff --git a/crates/api/src/routes/experiments.rs b/crates/api/src/routes/experiments.rs new file mode 100644 index 0000000..089d6f5 --- /dev/null +++ b/crates/api/src/routes/experiments.rs @@ -0,0 +1,470 @@ +//! A/B experiments. +//! +//! Environment-scoped like segments, and for a stronger reason: an experiment +//! *is* a measurement of one environment's traffic. Its lifecycle is the state +//! machine the storage layer enforces — draft, running, stopped, one way. + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use flagforge_core::VariantCounts; +use flagforge_storage::models::{Experiment, ExperimentState, NewAuditEntry}; +use flagforge_storage::{audit, experiments, flags, projects}; +use serde::{Deserialize, Serialize}; + +use crate::auth::AuthUser; +use crate::error::{ApiError, ApiResult}; +use crate::routes::valid_key; +use crate::state::AppState; + +#[derive(Debug, Deserialize, utoipa::ToSchema)] +pub struct CreateExperiment { + #[schema(example = "checkout-cta")] + pub key: String, + #[schema(example = "Checkout call-to-action")] + pub name: String, + pub description: Option, + /// The flag whose variants are being compared. + #[schema(example = "checkout.v2")] + pub flag_key: String, + /// Conversion events carrying this metric key count toward the experiment. + #[schema(example = "order.completed")] + pub metric_key: String, + /// The baseline variant; every other variant is judged against it. + #[schema(example = "off")] + pub control_variant: String, +} + +/// Presentation fields only. The measurement fields — flag, metric, control — +/// are fixed at creation: changing what a running experiment measures would +/// splice two questions into one answer, and a draft is cheap to recreate. +#[derive(Debug, Deserialize, utoipa::ToSchema)] +pub struct UpdateExperiment { + pub name: Option, + pub description: Option>, +} + +/// An experiment next to its judged results, one entry per variant. +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct ExperimentResults { + pub experiment: Experiment, + /// In the flag's variant order, zero-filled for arms nobody has reached, + /// so every arm renders even before its first exposure. A variant that + /// appears in the counters but not on the flag — possible after a variant + /// rename — is appended rather than hidden. + pub results: Vec, +} + +#[utoipa::path( + get, path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments", + tag = "experiments", security(("bearer" = [])), + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ), + responses((status = 200, description = "Experiments in this environment", body = Vec)) +)] +pub async fn list( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key)): Path<(String, String)>, +) -> ApiResult>> { + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + Ok(Json(experiments::list_experiments(&state.pool, environment.id).await?)) +} + +#[utoipa::path( + post, path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments", + tag = "experiments", security(("bearer" = [])), request_body = CreateExperiment, + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ), + responses( + (status = 201, description = "Experiment created as a draft", body = Experiment), + (status = 400, description = "Unknown flag, archived flag, or a control that is not one of its variants"), + (status = 409, description = "An experiment with that key already exists here"), + ) +)] +pub async fn create( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key)): Path<(String, String)>, + Json(body): Json, +) -> ApiResult<(StatusCode, Json)> { + caller.require_write()?; + valid_key(&body.key, "key")?; + valid_key(&body.metric_key, "metric_key")?; + + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + let flag = flags::find_flag(&state.pool, project.id, &body.flag_key).await?; + if flag.archived { + return Err(ApiError::BadRequest(format!( + "flag `{}` is archived; an experiment would measure traffic it no longer serves", + body.flag_key + ))); + } + if !flag.variants.iter().any(|v| v.key == body.control_variant) { + return Err(ApiError::BadRequest(format!( + "`{}` is not a variant of `{}` (it has: {})", + body.control_variant, + body.flag_key, + flag.variants.iter().map(|v| v.key.as_str()).collect::>().join(", ") + ))); + } + + let experiment = experiments::create_experiment( + &state.pool, + environment.id, + flag.id, + &body.key, + &body.name, + body.description.as_deref(), + &body.metric_key, + &body.control_variant, + ) + .await?; + + audit::record( + &state.pool, + NewAuditEntry::new( + caller.organization_id, + caller.actor(), + "experiment.created", + "experiment", + format!("{project_key}/{environment_key}/{}", body.key), + ) + .in_environment(environment.id) + .changing(None, Some(&experiment)), + ) + .await?; + + Ok((StatusCode::CREATED, Json(experiment))) +} + +#[utoipa::path( + get, + path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}", + tag = "experiments", security(("bearer" = [])), + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ("experiment_key" = String, Path, description = "Experiment key"), + ), + responses((status = 200, description = "The experiment", body = Experiment)) +)] +pub async fn get( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key, experiment_key)): Path<(String, String, String)>, +) -> ApiResult> { + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + Ok(Json(experiments::find_experiment(&state.pool, environment.id, &experiment_key).await?)) +} + +#[utoipa::path( + patch, + path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}", + tag = "experiments", security(("bearer" = [])), request_body = UpdateExperiment, + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ("experiment_key" = String, Path, description = "Experiment key"), + ), + responses((status = 200, description = "Experiment updated", body = Experiment)) +)] +pub async fn update( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key, experiment_key)): Path<(String, String, String)>, + Json(body): Json, +) -> ApiResult> { + caller.require_write()?; + + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + let previous = + experiments::find_experiment(&state.pool, environment.id, &experiment_key).await?; + let experiment = experiments::update_experiment( + &state.pool, + environment.id, + &experiment_key, + body.name.as_deref(), + body.description.as_ref().map(|d| d.as_deref()), + ) + .await?; + + audit::record( + &state.pool, + NewAuditEntry::new( + caller.organization_id, + caller.actor(), + "experiment.updated", + "experiment", + format!("{project_key}/{environment_key}/{experiment_key}"), + ) + .in_environment(environment.id) + .changing(Some(&previous), Some(&experiment)), + ) + .await?; + + Ok(Json(experiment)) +} + +#[utoipa::path( + post, + path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}/start", + tag = "experiments", security(("bearer" = [])), + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ("experiment_key" = String, Path, description = "Experiment key"), + ), + responses( + (status = 200, description = "Now running; SDKs will start recording", body = Experiment), + (status = 409, description = "Not a draft — a stopped experiment cannot restart"), + ) +)] +pub async fn start( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key, experiment_key)): Path<(String, String, String)>, +) -> ApiResult> { + transition( + state, + caller, + project_key, + environment_key, + experiment_key, + ExperimentState::Running, + ) + .await +} + +#[utoipa::path( + post, + path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}/stop", + tag = "experiments", security(("bearer" = [])), + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ("experiment_key" = String, Path, description = "Experiment key"), + ), + responses( + (status = 200, description = "Stopped; the results are final", body = Experiment), + (status = 409, description = "Only a running experiment can stop"), + ) +)] +pub async fn stop( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key, experiment_key)): Path<(String, String, String)>, +) -> ApiResult> { + transition( + state, + caller, + project_key, + environment_key, + experiment_key, + ExperimentState::Stopped, + ) + .await +} + +async fn transition( + state: AppState, + caller: AuthUser, + project_key: String, + environment_key: String, + experiment_key: String, + to: ExperimentState, +) -> ApiResult> { + caller.require_write()?; + + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + let (experiment, action) = match to { + ExperimentState::Running => ( + experiments::start_experiment(&state.pool, environment.id, &experiment_key).await?, + "experiment.started", + ), + _ => ( + experiments::stop_experiment(&state.pool, environment.id, &experiment_key).await?, + "experiment.stopped", + ), + }; + + // Starting and stopping both change what SDKs should record, and the + // snapshot is how they find out. Refreshing inline means the caller's next + // read sees its own write rather than racing the trigger's notification. + state.cache.refresh(environment.id).await; + + audit::record( + &state.pool, + NewAuditEntry::new( + caller.organization_id, + caller.actor(), + action, + "experiment", + format!("{project_key}/{environment_key}/{experiment_key}"), + ) + .in_environment(environment.id) + .changing(None, Some(&experiment)), + ) + .await?; + + Ok(Json(experiment)) +} + +#[utoipa::path( + delete, + path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}", + tag = "experiments", security(("bearer" = [])), + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ("experiment_key" = String, Path, description = "Experiment key"), + ), + responses( + (status = 204, description = "Experiment and its counters deleted"), + (status = 409, description = "Running — stop it first"), + ) +)] +pub async fn delete( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key, experiment_key)): Path<(String, String, String)>, +) -> ApiResult { + caller.require_write()?; + + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + let previous = + experiments::find_experiment(&state.pool, environment.id, &experiment_key).await?; + experiments::delete_experiment(&state.pool, environment.id, &experiment_key).await?; + + audit::record( + &state.pool, + NewAuditEntry::new( + caller.organization_id, + caller.actor(), + "experiment.deleted", + "experiment", + format!("{project_key}/{environment_key}/{experiment_key}"), + ) + .in_environment(environment.id) + .changing(Some(&previous), None), + ) + .await?; + + Ok(StatusCode::NO_CONTENT) +} + +#[utoipa::path( + get, + path = "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}/results", + tag = "experiments", security(("bearer" = [])), + params( + ("project_key" = String, Path, description = "Project key"), + ("environment_key" = String, Path, description = "Environment key"), + ("experiment_key" = String, Path, description = "Experiment key"), + ), + responses((status = 200, description = "Counts, rates, intervals and verdicts per variant", body = ExperimentResults)) +)] +pub async fn results( + State(state): State, + caller: AuthUser, + Path((project_key, environment_key, experiment_key)): Path<(String, String, String)>, +) -> ApiResult> { + let project = projects::find_project(&state.pool, caller.organization_id, &project_key).await?; + let environment = projects::find_environment(&state.pool, project.id, &environment_key).await?; + + let experiment = + experiments::find_experiment(&state.pool, environment.id, &experiment_key).await?; + let counts = experiments::experiment_counts(&state.pool, experiment.id).await?; + + let judged = + flagforge_core::results(&experiment.control_variant, &in_flag_order(&experiment, counts)); + + Ok(Json(ExperimentResults { experiment, results: judged })) +} + +/// Orders counts by the flag's variant list, zero-filling arms that have no +/// counters yet and appending any counter variant the flag no longer defines. +fn in_flag_order(experiment: &Experiment, mut counts: Vec) -> Vec { + let mut ordered = Vec::with_capacity(experiment.variants.len() + counts.len()); + for variant in &experiment.variants { + match counts.iter().position(|c| c.variant == variant.key) { + Some(index) => ordered.push(counts.swap_remove(index)), + None => ordered.push(VariantCounts { + variant: variant.key.clone(), + exposures: 0, + conversions: 0, + }), + } + } + counts.sort_by(|a, b| a.variant.cmp(&b.variant)); + ordered.extend(counts); + ordered +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::Utc; + use flagforge_core::Variant; + use uuid::Uuid; + + fn experiment() -> Experiment { + Experiment { + id: Uuid::nil(), + environment_id: Uuid::nil(), + flag_id: Uuid::nil(), + flag_key: "checkout.v2".into(), + variants: vec![Variant::new("on", true), Variant::new("off", false)], + key: "checkout-cta".into(), + name: "Checkout CTA".into(), + description: None, + metric_key: "order.completed".into(), + control_variant: "off".into(), + state: ExperimentState::Running, + started_at: None, + stopped_at: None, + version: 1, + created_at: Utc::now(), + updated_at: Utc::now(), + } + } + + fn counts(variant: &str, exposures: u64) -> VariantCounts { + VariantCounts { variant: variant.into(), exposures, conversions: 0 } + } + + #[test] + fn every_arm_renders_even_before_its_first_exposure() { + let ordered = in_flag_order(&experiment(), vec![counts("off", 10)]); + assert_eq!( + ordered.iter().map(|c| (c.variant.as_str(), c.exposures)).collect::>(), + [("on", 0), ("off", 10)], + "flag order, zero-filled" + ); + } + + #[test] + fn counters_for_a_renamed_variant_are_appended_not_hidden() { + let ordered = in_flag_order(&experiment(), vec![counts("legacy", 7), counts("on", 3)]); + assert_eq!( + ordered.iter().map(|c| c.variant.as_str()).collect::>(), + ["on", "off", "legacy"] + ); + } +} diff --git a/crates/api/src/routes/mod.rs b/crates/api/src/routes/mod.rs index 697321b..59a3157 100644 --- a/crates/api/src/routes/mod.rs +++ b/crates/api/src/routes/mod.rs @@ -3,6 +3,8 @@ pub mod audit; pub mod auth; pub mod evaluate; +pub mod events; +pub mod experiments; pub mod flags; pub mod health; pub mod keys; @@ -102,6 +104,26 @@ pub fn router(state: AppState, metrics: PrometheusHandle) -> Router { "/api/v1/projects/{project_key}/environments/{environment_key}/segments/{segment_key}", get(segments::get).put(segments::update).delete(segments::delete), ) + .route( + "/api/v1/projects/{project_key}/environments/{environment_key}/experiments", + get(experiments::list).post(experiments::create), + ) + .route( + "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}", + get(experiments::get).patch(experiments::update).delete(experiments::delete), + ) + .route( + "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}/start", + post(experiments::start), + ) + .route( + "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}/stop", + post(experiments::stop), + ) + .route( + "/api/v1/projects/{project_key}/environments/{environment_key}/experiments/{experiment_key}/results", + get(experiments::results), + ) .route( "/api/v1/projects/{project_key}/environments/{environment_key}/keys", get(keys::list).post(keys::create), @@ -119,6 +141,7 @@ pub fn router(state: AppState, metrics: PrometheusHandle) -> Router { .route("/api/v1/evaluate", post(evaluate::evaluate_all)) .route("/api/v1/evaluate/{flag_key}", post(evaluate::evaluate_one)) .route("/api/v1/snapshot", get(evaluate::snapshot)) + .route("/api/v1/events", post(events::ingest)) .layer( CorsLayer::new() .allow_origin(Any) diff --git a/crates/api/src/seed.rs b/crates/api/src/seed.rs index f9698a8..9f894d3 100644 --- a/crates/api/src/seed.rs +++ b/crates/api/src/seed.rs @@ -251,6 +251,8 @@ pub async fn run( .await?; } + seed_experiment(pool, organization.id, actor, project.id, &environments).await?; + // One key per environment, printed once — exactly as the dashboard does. let mut secrets = Vec::new(); for environment in &environments { @@ -271,6 +273,112 @@ pub async fn run( Ok(()) } +/// One experiment on `checkout.v2` per environment: running with a week of +/// counters in production, still a draft in staging — the two states an +/// operator actually encounters, on the first screen they open. +/// +/// The counters are deterministic rather than random (seeding must be +/// reproducible), with a small modular wobble so the hourly cells do not all +/// carry the same number: the treatment converts at ~13%, the control at ~9%, +/// enough for the z-test to reach a verdict on a week of data. +async fn seed_experiment( + pool: &PgPool, + organization_id: Uuid, + actor: (Option, &str), + project_id: Uuid, + environments: &[flagforge_storage::models::Environment], +) -> anyhow::Result<()> { + use chrono::{TimeDelta, Utc}; + use flagforge_storage::experiments; + use flagforge_storage::models::{CounterDelta, CounterKind}; + + const EXPERIMENT_KEY: &str = "checkout-cta"; + const HOURS: i64 = 24 * 7; + + let flag = flags::find_flag(pool, project_id, "checkout.v2").await?; + + for environment in environments { + experiments::create_experiment( + pool, + environment.id, + flag.id, + EXPERIMENT_KEY, + "Checkout call-to-action", + Some("Does the rebuilt checkout convert better than the old one?"), + "order.completed", + "off", + ) + .await?; + + audit::record( + pool, + NewAuditEntry::new( + organization_id, + actor, + "experiment.created", + "experiment", + format!("checkout/{}/{EXPERIMENT_KEY}", environment.key), + ) + .in_environment(environment.id), + ) + .await?; + + if !environment.is_production { + continue; + } + + experiments::start_experiment(pool, environment.id, EXPERIMENT_KEY).await?; + + let now = Utc::now(); + let mut deltas = Vec::new(); + for hour in 0..HOURS { + let at = now - TimeDelta::hours(hour); + let exposures = 40 + (hour % 9) as u32; + for (variant, conversions) in + [("on", 5 + (hour % 3) as u32), ("off", 3 + (hour % 3) as u32)] + { + deltas.push(CounterDelta { + experiment_key: EXPERIMENT_KEY.to_owned(), + variant: variant.to_owned(), + kind: CounterKind::Exposure, + at, + count: exposures, + }); + deltas.push(CounterDelta { + experiment_key: EXPERIMENT_KEY.to_owned(), + variant: variant.to_owned(), + kind: CounterKind::Conversion, + at, + count: conversions, + }); + } + } + experiments::ingest_counters(pool, environment.id, &deltas).await?; + experiments::backdate_start( + pool, + environment.id, + EXPERIMENT_KEY, + now - TimeDelta::hours(HOURS), + ) + .await?; + + audit::record( + pool, + NewAuditEntry::new( + organization_id, + actor, + "experiment.started", + "experiment", + format!("checkout/{}/{EXPERIMENT_KEY}", environment.key), + ) + .in_environment(environment.id), + ) + .await?; + } + + Ok(()) +} + fn report(email: &str, owner_password: &str, generated: bool, secrets: &[(String, String)]) { println!("\nSeeded a demo organization.\n"); println!(" Sign in at /"); diff --git a/crates/api/tests/experiments.rs b/crates/api/tests/experiments.rs new file mode 100644 index 0000000..0adc14d --- /dev/null +++ b/crates/api/tests/experiments.rs @@ -0,0 +1,242 @@ +//! Experiments end to end: lifecycle, event ingest, judged results, guardrails. + +mod harness; + +use axum::http::StatusCode; +use harness::TestApp; +use serde_json::{Value, json}; +use sqlx::PgPool; + +const EXPERIMENTS: &str = "/api/v1/projects/checkout/environments/production/experiments"; + +fn checkout_experiment() -> Value { + json!({ + "key": "checkout-cta", + "name": "Checkout CTA", + "flag_key": "checkout.v2", + "metric_key": "order.completed", + "control_variant": "off", + }) +} + +/// A batch of `count`-weighted events for one variant. +fn events(variant: &str, exposures: u64, conversions: u64) -> Value { + json!({ + "events": [ + {"experiment_key": "checkout-cta", "variant": variant, + "kind": "exposure", "count": exposures}, + {"experiment_key": "checkout-cta", "variant": variant, + "kind": "conversion", "count": conversions}, + ] + }) +} + +#[sqlx::test(migrations = "../../migrations")] +async fn the_full_lifecycle_measures_and_stops_measuring(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + app.create_flag(&tenant.token, "checkout.v2").await; + + app.post(EXPERIMENTS, Some(&tenant.token), checkout_experiment()) + .await + .expect(StatusCode::CREATED); + + // A draft is not yet measuring, so the snapshot must not announce it. + let snapshot = app.get("/api/v1/snapshot", Some(&tenant.sdk_key)).await.expect(StatusCode::OK); + assert_eq!(snapshot["experiments"], json!([]), "drafts must not reach SDKs"); + + // Events for a draft are dropped, not errors: an SDK cannot know yet. + let dropped = app + .post("/api/v1/events", Some(&tenant.sdk_key), events("on", 10, 1)) + .await + .expect(StatusCode::ACCEPTED); + assert_eq!(dropped["accepted"], 0); + assert_eq!(dropped["received"], 2); + + app.post(&format!("{EXPERIMENTS}/checkout-cta/start"), Some(&tenant.token), json!({})) + .await + .expect(StatusCode::OK); + + let snapshot = app.get("/api/v1/snapshot", Some(&tenant.sdk_key)).await.expect(StatusCode::OK); + assert_eq!(snapshot["experiments"][0]["key"], "checkout-cta"); + assert_eq!(snapshot["experiments"][0]["flag_key"], "checkout.v2"); + assert_eq!(snapshot["experiments"][0]["metric_key"], "order.completed"); + + // A clear difference: 15% converting against 10% over a thousand each. + for (variant, exposures, conversions) in [("on", 1000, 150), ("off", 1000, 100)] { + app.post("/api/v1/events", Some(&tenant.sdk_key), events(variant, exposures, conversions)) + .await + .expect(StatusCode::ACCEPTED); + } + + let body = app + .get(&format!("{EXPERIMENTS}/checkout-cta/results"), Some(&tenant.token)) + .await + .expect(StatusCode::OK); + + let results = body["results"].as_array().expect("results array"); + assert_eq!(results.len(), 2, "{body}"); + let on = &results[0]; + assert_eq!(on["variant"], "on"); + assert_eq!(on["exposures"], 1000); + assert_eq!(on["conversions"], 150); + assert_eq!(on["rate"], 0.15); + assert_eq!(on["vs_control"]["significant"], true, "{on}"); + assert!(on["vs_control"]["p_value"].as_f64().unwrap() < 0.001); + // The control carries numbers but no verdict about itself. + assert_eq!(results[1]["variant"], "off"); + assert!(results[1]["vs_control"].is_null()); + + app.post(&format!("{EXPERIMENTS}/checkout-cta/stop"), Some(&tenant.token), json!({})) + .await + .expect(StatusCode::OK); + + // Stopped: gone from the snapshot, and late events no longer count. + let snapshot = app.get("/api/v1/snapshot", Some(&tenant.sdk_key)).await.expect(StatusCode::OK); + assert_eq!(snapshot["experiments"], json!([])); + let late = app + .post("/api/v1/events", Some(&tenant.sdk_key), events("on", 5, 1)) + .await + .expect(StatusCode::ACCEPTED); + assert_eq!(late["accepted"], 0, "a stopped experiment must stop counting"); + + // And stopped is terminal. + let refused = app + .post(&format!("{EXPERIMENTS}/checkout-cta/start"), Some(&tenant.token), json!({})) + .await; + refused.expect(StatusCode::CONFLICT); +} + +#[sqlx::test(migrations = "../../migrations")] +async fn every_arm_renders_before_any_event_arrives(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + app.create_flag(&tenant.token, "checkout.v2").await; + + app.post(EXPERIMENTS, Some(&tenant.token), checkout_experiment()) + .await + .expect(StatusCode::CREATED); + + let body = app + .get(&format!("{EXPERIMENTS}/checkout-cta/results"), Some(&tenant.token)) + .await + .expect(StatusCode::OK); + + let results = body["results"].as_array().expect("results array"); + assert_eq!(results.len(), 2, "both arms must render zero-filled: {body}"); + for arm in results { + assert_eq!(arm["exposures"], 0); + assert!(arm["rate"].is_null(), "no exposures means no rate, not 0%"); + assert!(arm["vs_control"].is_null()); + } +} + +#[sqlx::test(migrations = "../../migrations")] +async fn a_control_that_is_not_a_variant_is_refused(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + app.create_flag(&tenant.token, "checkout.v2").await; + + let mut body = checkout_experiment(); + body["control_variant"] = "ghost".into(); + let refused = app.post(EXPERIMENTS, Some(&tenant.token), body).await; + let problem = refused.expect(StatusCode::BAD_REQUEST); + // The message must name the variants that would have been accepted. + assert!(problem["title"].as_str().unwrap().contains("on, off"), "{problem}"); +} + +#[sqlx::test(migrations = "../../migrations")] +async fn an_experiment_pins_its_flag_against_deletion(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + app.create_flag(&tenant.token, "checkout.v2").await; + + app.post(EXPERIMENTS, Some(&tenant.token), checkout_experiment()) + .await + .expect(StatusCode::CREATED); + + let refused = + app.delete("/api/v1/projects/checkout/flags/checkout.v2", Some(&tenant.token)).await; + let problem = refused.expect(StatusCode::CONFLICT); + assert!( + problem["title"].as_str().unwrap().contains("production/checkout-cta"), + "the refusal must name the experiment: {problem}" + ); + + // Removing the experiment unpins the flag. + app.delete(&format!("{EXPERIMENTS}/checkout-cta"), Some(&tenant.token)) + .await + .expect(StatusCode::NO_CONTENT); + app.delete("/api/v1/projects/checkout/flags/checkout.v2", Some(&tenant.token)) + .await + .expect(StatusCode::NO_CONTENT); +} + +#[sqlx::test(migrations = "../../migrations")] +async fn a_running_experiment_cannot_be_deleted_mid_measurement(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + app.create_flag(&tenant.token, "checkout.v2").await; + + app.post(EXPERIMENTS, Some(&tenant.token), checkout_experiment()) + .await + .expect(StatusCode::CREATED); + app.post(&format!("{EXPERIMENTS}/checkout-cta/start"), Some(&tenant.token), json!({})) + .await + .expect(StatusCode::OK); + + app.delete(&format!("{EXPERIMENTS}/checkout-cta"), Some(&tenant.token)) + .await + .expect(StatusCode::CONFLICT); + + app.post(&format!("{EXPERIMENTS}/checkout-cta/stop"), Some(&tenant.token), json!({})) + .await + .expect(StatusCode::OK); + app.delete(&format!("{EXPERIMENTS}/checkout-cta"), Some(&tenant.token)) + .await + .expect(StatusCode::NO_CONTENT); +} + +#[sqlx::test(migrations = "../../migrations")] +async fn duplicate_cells_in_one_batch_are_summed_not_a_server_error(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + app.create_flag(&tenant.token, "checkout.v2").await; + + app.post(EXPERIMENTS, Some(&tenant.token), checkout_experiment()) + .await + .expect(StatusCode::CREATED); + app.post(&format!("{EXPERIMENTS}/checkout-cta/start"), Some(&tenant.token), json!({})) + .await + .expect(StatusCode::OK); + + // Three deltas landing in the same hourly cell — the shape that makes a + // naive multi-row upsert fail with "cannot affect row a second time". + let body = json!({ + "events": [ + {"experiment_key": "checkout-cta", "variant": "on", "kind": "exposure", "count": 1}, + {"experiment_key": "checkout-cta", "variant": "on", "kind": "exposure", "count": 2}, + {"experiment_key": "checkout-cta", "variant": "on", "kind": "exposure", "count": 3}, + ] + }); + app.post("/api/v1/events", Some(&tenant.sdk_key), body).await.expect(StatusCode::ACCEPTED); + + let results = app + .get(&format!("{EXPERIMENTS}/checkout-cta/results"), Some(&tenant.token)) + .await + .expect(StatusCode::OK); + assert_eq!(results["results"][0]["exposures"], 6, "{results}"); +} + +#[sqlx::test(migrations = "../../migrations")] +async fn events_require_an_sdk_key(pool: PgPool) { + let app = TestApp::new(pool); + let tenant = app.bootstrap("Acme", "ada@acme.test").await; + + app.post("/api/v1/events", None, events("on", 1, 0)).await.expect(StatusCode::UNAUTHORIZED); + + // A management token is not an SDK key either. + app.post("/api/v1/events", Some(&tenant.token), events("on", 1, 0)) + .await + .expect(StatusCode::UNAUTHORIZED); +} diff --git a/crates/core/src/experiment.rs b/crates/core/src/experiment.rs new file mode 100644 index 0000000..25c5754 --- /dev/null +++ b/crates/core/src/experiment.rs @@ -0,0 +1,323 @@ +//! Experiments: measuring what a flag's variants do to a metric. +//! +//! An experiment binds a flag to a conversion metric. Assignment reuses the +//! engine's deterministic bucketing — the same hash that makes a rollout sticky +//! makes an experiment's cohorts stable — so this module only has to describe +//! experiments and judge their numbers. Both halves are pure: the spec is data, +//! and the statistics are functions from counts to conclusions, testable +//! without a database or a clock. + +use serde::{Deserialize, Serialize}; + +/// Confidence level used for intervals and the significance threshold, as the +/// conventional two-sided 95 %. +pub const ALPHA: f64 = 0.05; + +/// z-score for a two-sided 95 % interval. +const Z_95: f64 = 1.959_963_984_540_054; + +/// The part of an experiment an SDK needs: enough to attribute events, nothing +/// it could misuse. Travels in the environment snapshot while the experiment is +/// running. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct ExperimentSpec { + pub key: String, + /// The flag whose variants are being compared. + pub flag_key: String, + /// Conversion events with this metric key count toward the experiment. + pub metric_key: String, + /// The baseline variant every other variant is compared against. + pub control_variant: String, + /// Monotonic counter bumped on every write, folded into the snapshot + /// version so starting or stopping an experiment is a visible change. + #[serde(default)] + pub version: i64, +} + +/// Raw tallies for one variant, as stored: exposures are evaluations of the +/// flag while the experiment ran, conversions are tracked metric events. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct VariantCounts { + pub variant: String, + pub exposures: u64, + pub conversions: u64, +} + +/// A 95 % Wilson score interval around a conversion rate. +/// +/// Wilson rather than the normal approximation because experiments spend their +/// early life exactly where the normal interval misbehaves: small samples and +/// rates near zero, where it happily reports negative lower bounds. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct ConfidenceInterval { + pub low: f64, + pub high: f64, +} + +/// A variant measured against the control. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct Comparison { + /// Absolute difference in conversion rate, variant minus control. + pub lift: f64, + /// Two-proportion z statistic, pooled variance. + pub z: f64, + /// Two-sided p-value for "the rates are equal". + pub p_value: f64, + /// Whether `p_value` clears [`ALPHA`]. Stored rather than recomputed so a + /// reader of the wire format sees the verdict next to the evidence. + pub significant: bool, +} + +/// Everything the dashboard shows about one variant. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))] +pub struct VariantResult { + pub variant: String, + pub exposures: u64, + pub conversions: u64, + /// Conversion rate, absent until the variant has an exposure. Conversions + /// beyond the exposure count — possible when a metric is tracked for a + /// context that never evaluated the flag — are clamped here but reported + /// raw above, so the mismatch is visible instead of laundered. + pub rate: Option, + pub interval: Option, + /// Absent for the control itself and while either side has no exposures. + pub vs_control: Option, +} + +/// Judges every variant's counts against the control's. +/// +/// Variants come back in input order. A control that is missing from `counts` +/// (or has no exposures yet) yields results with every comparison absent — +/// the numbers still render, the verdicts wait for data. +pub fn results(control_variant: &str, counts: &[VariantCounts]) -> Vec { + let control = counts.iter().find(|c| c.variant == control_variant).and_then(observed); + + counts + .iter() + .map(|c| { + let own = observed(c); + let vs_control = match (own, control) { + // Comparing the control to itself would report certainty about + // nothing, so it gets no comparison rather than a perfect one. + _ if c.variant == control_variant => None, + (Some(own), Some(control)) => two_proportion_z(own, control), + _ => None, + }; + + VariantResult { + variant: c.variant.clone(), + exposures: c.exposures, + conversions: c.conversions, + rate: own.map(|(x, n)| x / n), + interval: own.map(|(x, n)| wilson(x / n, n)), + vs_control, + } + }) + .collect() +} + +/// Clamped (conversions, exposures) as floats, or `None` without exposures. +fn observed(c: &VariantCounts) -> Option<(f64, f64)> { + (c.exposures > 0).then(|| (c.conversions.min(c.exposures) as f64, c.exposures as f64)) +} + +/// 95 % Wilson score interval for `rate` observed over `n` trials. +fn wilson(rate: f64, n: f64) -> ConfidenceInterval { + let z2 = Z_95 * Z_95; + let denominator = 1.0 + z2 / n; + let center = (rate + z2 / (2.0 * n)) / denominator; + let half_width = (Z_95 / denominator) * (rate * (1.0 - rate) / n + z2 / (4.0 * n * n)).sqrt(); + + // The clamps also absorb float rounding at the boundaries: at an observed + // rate of exactly 1 the algebra gives `high = 1` but the arithmetic can + // land one ulp below, and an interval excluding its own point estimate is + // a downstream rendering bug waiting to happen. + ConfidenceInterval { + low: (center - half_width).clamp(0.0, rate), + high: (center + half_width).clamp(rate, 1.0), + } +} + +/// Pooled two-proportion z-test of `(x, n)` against the control's `(x, n)`. +/// +/// Returns `None` when the pooled rate is 0 or 1 — the standard error is zero +/// there, and with no observed variance the test has nothing to say. +fn two_proportion_z(own: (f64, f64), control: (f64, f64)) -> Option { + let (x1, n1) = own; + let (x2, n2) = control; + + let pooled = (x1 + x2) / (n1 + n2); + let variance = pooled * (1.0 - pooled) * (1.0 / n1 + 1.0 / n2); + if variance <= 0.0 { + return None; + } + + let lift = x1 / n1 - x2 / n2; + let z = lift / variance.sqrt(); + let p_value = 2.0 * (1.0 - phi(z.abs())); + + Some(Comparison { lift, z, p_value, significant: p_value < ALPHA }) +} + +/// Standard normal CDF. +fn phi(x: f64) -> f64 { + 0.5 * (1.0 + erf(x / std::f64::consts::SQRT_2)) +} + +/// Error function via Abramowitz & Stegun 7.1.26. +/// +/// Hand-rolled on purpose: the alternative is a statistics crate an order of +/// magnitude larger than this module, and the approximation's maximum absolute +/// error (1.5 × 10⁻⁷) is noise four decimal places below any p-value cutoff a +/// dashboard reader would act on. +fn erf(x: f64) -> f64 { + let sign = if x < 0.0 { -1.0 } else { 1.0 }; + let x = x.abs(); + + let t = 1.0 / (1.0 + 0.327_591_1 * x); + let polynomial = t + * (0.254_829_592 + + t * (-0.284_496_736 + + t * (1.421_413_741 + t * (-1.453_152_027 + t * 1.061_405_429)))); + + sign * (1.0 - polynomial * (-x * x).exp()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn counts(variant: &str, exposures: u64, conversions: u64) -> VariantCounts { + VariantCounts { variant: variant.into(), exposures, conversions } + } + + #[test] + fn the_spec_wire_format_is_stable() { + let spec = ExperimentSpec { + key: "checkout-cta".into(), + flag_key: "checkout.v2".into(), + metric_key: "checkout.completed".into(), + control_variant: "off".into(), + version: 3, + }; + let json = serde_json::to_string(&spec).unwrap(); + assert_eq!( + json, + r#"{"key":"checkout-cta","flag_key":"checkout.v2","metric_key":"checkout.completed","control_variant":"off","version":3}"# + ); + // And a payload from before `version` existed still deserializes. + let old = r#"{"key":"k","flag_key":"f","metric_key":"m","control_variant":"off"}"#; + assert_eq!(serde_json::from_str::(old).unwrap().version, 0); + } + + #[test] + fn erf_matches_published_values() { + // Abramowitz & Stegun table values; the approximation is good to 1.5e-7. + for (x, expected) in [(0.5, 0.520_499_878), (1.0, 0.842_700_793), (2.0, 0.995_322_265)] { + assert!((erf(x) - expected).abs() < 1e-6, "erf({x}) = {}", erf(x)); + assert!((erf(-x) + expected).abs() < 1e-6, "erf must be odd"); + } + } + + #[test] + fn a_real_difference_is_called_significant() { + // 10 % vs 15 % over a thousand exposures each: z ≈ 3.38, p ≈ 0.0007. + let out = results("control", &[counts("control", 1000, 100), counts("treat", 1000, 150)]); + + let treat = &out[1]; + let cmp = treat.vs_control.unwrap(); + assert!((cmp.lift - 0.05).abs() < 1e-12); + assert!((3.3..3.5).contains(&cmp.z), "z = {}", cmp.z); + assert!(cmp.p_value < 0.001); + assert!(cmp.significant); + } + + #[test] + fn identical_rates_are_not_significant() { + let out = results("a", &[counts("a", 500, 50), counts("b", 500, 50)]); + let cmp = out[1].vs_control.unwrap(); + assert_eq!(cmp.lift, 0.0); + // Tolerance bounded by the erf approximation's error, not by ulps. + assert!((cmp.p_value - 1.0).abs() < 1e-6); + assert!(!cmp.significant); + } + + #[test] + fn the_comparison_is_antisymmetric() { + let a = counts("a", 800, 120); + let b = counts("b", 900, 90); + let ab = results("b", &[a.clone(), b.clone()])[0].vs_control.unwrap(); + let ba = results("a", &[a, b])[1].vs_control.unwrap(); + + assert!((ab.z + ba.z).abs() < 1e-12); + assert!((ab.p_value - ba.p_value).abs() < 1e-12); + } + + #[test] + fn the_control_gets_numbers_but_no_verdict_about_itself() { + let out = results("control", &[counts("control", 100, 30)]); + assert_eq!(out[0].rate, Some(0.3)); + assert!(out[0].interval.is_some()); + assert!(out[0].vs_control.is_none()); + } + + #[test] + fn without_exposures_everything_waits_for_data() { + let out = results("control", &[counts("control", 0, 0), counts("treat", 50, 5)]); + + assert_eq!(out[0].rate, None); + assert_eq!(out[0].interval, None); + // The treatment has data, but there is nothing to compare against. + assert_eq!(out[1].rate, Some(0.1)); + assert!(out[1].vs_control.is_none()); + } + + #[test] + fn a_missing_control_disables_comparisons_not_results() { + let out = results("ghost", &[counts("treat", 50, 5)]); + assert_eq!(out[0].rate, Some(0.1)); + assert!(out[0].vs_control.is_none()); + } + + #[test] + fn excess_conversions_are_reported_raw_but_clamped_in_the_rate() { + // Trackable without evaluating: a conversion can outnumber exposures. + let out = results("c", &[counts("c", 10, 25)]); + assert_eq!(out[0].conversions, 25, "the raw count must stay visible"); + assert_eq!(out[0].rate, Some(1.0)); + let interval = out[0].interval.unwrap(); + assert!(interval.high <= 1.0); + } + + #[test] + fn wilson_intervals_stay_inside_the_unit_range_and_cover_the_rate() { + for (n, x) in [(1_u64, 0_u64), (1, 1), (10, 0), (10, 10), (20, 1), (1000, 500)] { + let out = results("only", &[counts("only", n, x)]); + let rate = out[0].rate.unwrap(); + let interval = out[0].interval.unwrap(); + assert!( + (0.0..=1.0).contains(&interval.low) && (0.0..=1.0).contains(&interval.high), + "n={n} x={x}: {interval:?}" + ); + assert!( + interval.low <= rate && rate <= interval.high, + "n={n} x={x}: {rate} outside {interval:?}" + ); + } + } + + #[test] + fn all_or_nothing_pooled_rates_yield_no_verdict() { + // Nobody converted anywhere: zero variance, so no test. + let out = results("a", &[counts("a", 100, 0), counts("b", 100, 0)]); + assert!(out[1].vs_control.is_none()); + // Everybody converted everywhere: same. + let out = results("a", &[counts("a", 100, 100), counts("b", 100, 100)]); + assert!(out[1].vs_control.is_none()); + } +} diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index da372b1..7e67809 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -38,6 +38,7 @@ pub mod bucket; pub mod context; pub mod engine; +pub mod experiment; pub mod flag; pub mod matcher; pub mod segment; @@ -48,6 +49,9 @@ pub mod value; pub use bucket::{bucket, pick_weighted}; pub use context::EvaluationContext; pub use engine::{Evaluation, EvaluationEnv, Reason, evaluate}; +pub use experiment::{ + Comparison, ConfidenceInterval, ExperimentSpec, VariantCounts, VariantResult, results, +}; pub use flag::{ Condition, Distribution, Flag, Operator, Rule, TOTAL_WEIGHT, Variant, WeightedVariant, }; diff --git a/crates/core/src/snapshot.rs b/crates/core/src/snapshot.rs index 8cf7f6f..d999a3a 100644 --- a/crates/core/src/snapshot.rs +++ b/crates/core/src/snapshot.rs @@ -14,6 +14,7 @@ use uuid::Uuid; use crate::context::EvaluationContext; use crate::engine::{Evaluation, EvaluationEnv, evaluate}; +use crate::experiment::ExperimentSpec; use crate::flag::Flag; use crate::segment::{Segment, SegmentSet}; use crate::value::VariantValue; @@ -34,9 +35,16 @@ pub struct EnvironmentSnapshot { /// holding flags but no segments would quietly stop matching. #[serde(default)] pub segments: SegmentSet, - /// Highest `version` across flags *and* segments, used as an ETag-style - /// cache token. Segments count because editing one changes what every flag - /// referencing it serves, without touching any flag's own version. + /// The experiments currently running, so an SDK can attribute exposures + /// and conversions locally. Only running ones travel: a draft is not yet + /// measuring and a stopped one must stop counting everywhere at once. + #[serde(default)] + pub experiments: Vec, + /// Highest `version` across flags, segments *and* running experiments, + /// used as an ETag-style cache token. Segments count because editing one + /// changes what every flag referencing it serves, without touching any + /// flag's own version; experiments count because starting or stopping one + /// changes what SDKs should record. pub version: i64, pub generated_at: DateTime, } @@ -66,11 +74,21 @@ impl EnvironmentSnapshot { salt: salt.into(), flags, segments, + experiments: Vec::new(), version, generated_at, } } + /// Attaches the running experiments, folding their versions into the + /// snapshot's so starting or stopping one is a visible change. + pub fn with_experiments(mut self, experiments: Vec) -> Self { + self.version = + experiments.iter().map(|e| e.version).chain([self.version]).max().unwrap_or(0); + self.experiments = experiments; + self + } + pub fn get(&self, flag_key: &str) -> Option<&Flag> { self.flags.get(flag_key) } @@ -204,6 +222,34 @@ mod tests { assert!(!snap.evaluate("f", &free, VariantValue::Bool(false)).is_on()); } + /// Starting an experiment must move the snapshot version for the same + /// reason a segment edit does: SDK behaviour changes without any flag + /// having been written. + #[test] + fn version_covers_running_experiments_too() { + use crate::experiment::ExperimentSpec; + + let snap = + snapshot(vec![Flag { version: 3, ..Flag::boolean("a") }]).with_experiments(vec![ + ExperimentSpec { + key: "exp".into(), + flag_key: "a".into(), + metric_key: "converted".into(), + control_variant: "off".into(), + version: 8, + }, + ]); + assert_eq!(snap.version, 8); + assert_eq!(snap.experiments.len(), 1); + + // And a snapshot serialized before experiments existed still loads. + let old = r#"{"environment_id":"00000000-0000-0000-0000-000000000000", + "environment_key":"production","flags":{},"version":1, + "generated_at":"2026-01-01T00:00:00Z"}"#; + let parsed: EnvironmentSnapshot = serde_json::from_str(old).unwrap(); + assert!(parsed.experiments.is_empty()); + } + #[test] fn the_salt_never_leaks_into_serialized_output() { let snap = diff --git a/crates/sdk/src/client.rs b/crates/sdk/src/client.rs index 9e56d5c..ea76e38 100644 --- a/crates/sdk/src/client.rs +++ b/crates/sdk/src/client.rs @@ -8,6 +8,7 @@ use flagforge_core::{EnvironmentSnapshot, Evaluation, EvaluationContext, Variant use tokio::task::JoinHandle; use crate::error::Error; +use crate::events::{Cell, EventKind, EventPayload, Recorder}; /// How often the configuration is re-fetched by default. /// @@ -17,6 +18,12 @@ use crate::error::Error; /// [`Client::refresh`] wired to your own signal. const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(30); const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5); +/// How often accumulated experiment events are shipped by default. Faster than +/// the poll: counters are tiny, and a short interval keeps a deploy's final +/// unflushed window small. +const DEFAULT_EVENT_FLUSH_INTERVAL: Duration = Duration::from_secs(10); +/// The server refuses batches above this; the flusher chunks to it. +const MAX_EVENTS_PER_BATCH: usize = 1_000; /// Backoff cap for a server that is down; low enough to recover promptly, /// high enough not to add load to something already struggling. const MAX_BACKOFF: Duration = Duration::from_secs(60); @@ -28,10 +35,13 @@ pub type Snapshot = EnvironmentSnapshot; struct Inner { http: reqwest::Client, url: String, + events_url: String, key: String, /// `None` until the first successful fetch. Lock-free to read, because /// every flag check reads it. current: ArcSwapOption, + /// Experiment exposures and conversions waiting for the next flush. + recorder: Recorder, } impl Inner { @@ -61,6 +71,68 @@ impl Inner { fn store(&self, snapshot: Snapshot) { self.current.store(Some(Arc::new(snapshot))); } + + /// Ships accumulated event counts, in server-sized chunks. + /// + /// On any failure the unsent counts go back into the recorder, so the next + /// flush retries them merged with whatever arrived in between — an outage + /// delays counters, it does not lose them (until the process exits). + async fn flush_events(&self) -> Result<(), Error> { + let (counts, dropped) = self.recorder.drain(); + if dropped > 0 { + tracing::warn!( + dropped, + "flagforge: experiment events dropped by the in-process cell cap" + ); + } + if counts.is_empty() { + return Ok(()); + } + + let cells: Vec<(Cell, u64)> = counts.into_iter().collect(); + for start in (0..cells.len()).step_by(MAX_EVENTS_PER_BATCH) { + let chunk = &cells[start..(start + MAX_EVENTS_PER_BATCH).min(cells.len())]; + let events: Vec> = chunk + .iter() + .map(|(cell, count)| EventPayload { + experiment_key: &cell.experiment_key, + variant: &cell.variant, + kind: cell.kind, + // Saturating: 4 billion identical events between two + // flushes is not a workload, it is a bug elsewhere. + count: u32::try_from(*count).unwrap_or(u32::MAX), + }) + .collect(); + + if let Err(error) = self.post_events(&events).await { + self.recorder.restore(cells[start..].iter().cloned().collect()); + return Err(error); + } + } + Ok(()) + } + + async fn post_events(&self, events: &[EventPayload<'_>]) -> Result<(), Error> { + let response = self + .http + .post(&self.events_url) + .header("authorization", format!("Bearer {}", self.key)) + .json(&serde_json::json!({ "events": events })) + .send() + .await?; + + let status = response.status().as_u16(); + if status == 401 || status == 403 { + return Err(Error::Unauthorized(detail(&response.text().await.unwrap_or_default()))); + } + if !(200..300).contains(&status) { + return Err(Error::Api { + status, + detail: detail(&response.text().await.unwrap_or_default()), + }); + } + Ok(()) + } } /// Pulls the human-readable part out of an RFC 9457 problem document. @@ -77,6 +149,7 @@ pub struct ClientBuilder { base_url: String, key: String, poll_interval: Duration, + event_flush_interval: Duration, request_timeout: Duration, } @@ -88,6 +161,14 @@ impl ClientBuilder { self } + /// How often accumulated experiment events are shipped. Pass + /// [`Duration::ZERO`] to disable the background flusher and call + /// [`Client::flush_events`] yourself. + pub fn event_flush_interval(mut self, interval: Duration) -> Self { + self.event_flush_interval = interval; + self + } + pub fn request_timeout(mut self, timeout: Duration) -> Self { self.request_timeout = timeout; self @@ -138,14 +219,22 @@ impl ClientBuilder { let inner = Arc::new(Inner { http, url: format!("{base}/api/v1/snapshot"), + events_url: format!("{base}/api/v1/events"), key: self.key, current: ArcSwapOption::empty(), + recorder: Recorder::default(), }); - let refresher = (!self.poll_interval.is_zero()) - .then(|| spawn_refresher(Arc::clone(&inner), self.poll_interval)); + let mut tasks = Vec::new(); + if !self.poll_interval.is_zero() { + tasks.push(spawn_refresher(Arc::clone(&inner), self.poll_interval)); + } + if !self.event_flush_interval.is_zero() { + tasks.push(spawn_event_flusher(Arc::clone(&inner), self.event_flush_interval)); + } + let tasks = (!tasks.is_empty()).then(|| Arc::new(tasks)); - Ok((Client { inner: Arc::clone(&inner), refresher: refresher.map(Arc::new) }, inner)) + Ok((Client { inner: Arc::clone(&inner), tasks }, inner)) } } @@ -153,7 +242,7 @@ impl ClientBuilder { #[derive(Debug, Clone)] pub struct Client { inner: Arc, - refresher: Option>>, + tasks: Option>>>, } impl Client { @@ -164,6 +253,7 @@ impl Client { base_url: base_url.into(), key: key.into(), poll_interval: DEFAULT_POLL_INTERVAL, + event_flush_interval: DEFAULT_EVENT_FLUSH_INTERVAL, request_timeout: DEFAULT_TIMEOUT, } } @@ -230,12 +320,68 @@ impl Client { fallback: VariantValue, ) -> Evaluation { match self.inner.current.load().as_ref() { - Some(snapshot) => snapshot.evaluate(flag, context, fallback), + Some(snapshot) => { + let evaluation = snapshot.evaluate(flag, context, fallback); + // A decision under a running experiment is an exposure — the + // denominator of that experiment's conversion rate. + if let Some(variant) = &evaluation.variant { + for experiment in &snapshot.experiments { + if experiment.flag_key == flag { + self.inner.recorder.record( + &experiment.key, + variant, + EventKind::Exposure, + ); + } + } + } + evaluation + } // Not loaded yet: behave exactly as if the flag did not exist. None => Evaluation::not_found(flag, fallback), } } + /// Records a conversion for every running experiment measuring `metric`. + /// + /// The variant is attributed by evaluating each experiment's flag for this + /// context — the same deterministic bucketing the exposure went through, + /// so both sides of the rate land on the same arm without the SDK keeping + /// any per-user state. The price of that statelessness: a conversion + /// counts even for a context that never actually evaluated the flag, and + /// the results view reports such excess rather than hiding it. + /// + /// Returns how many experiments counted the event — `0` means no running + /// experiment measures this metric, which is fine: instrument the metric + /// once and experiments can come and go without code changes. + pub fn track(&self, metric: &str, context: &EvaluationContext) -> usize { + let Some(snapshot) = self.inner.current.load_full() else { return 0 }; + + let mut credited = 0; + for experiment in &snapshot.experiments { + if experiment.metric_key != metric { + continue; + } + // Straight to the snapshot, not through `evaluate` — attributing a + // conversion must not also count an exposure. + let evaluation = snapshot.evaluate(&experiment.flag_key, context, VariantValue::null()); + if let Some(variant) = &evaluation.variant { + self.inner.recorder.record(&experiment.key, variant, EventKind::Conversion); + credited += 1; + } + } + credited + } + + /// Ships accumulated experiment events now. + /// + /// The background flusher does this on an interval; call it directly + /// before a graceful shutdown so the final window of counts is not lost + /// with the process. + pub async fn flush_events(&self) -> Result<(), Error> { + self.inner.flush_events().await + } + /// Every flag in the environment, for bulk logging or a debug endpoint. pub fn evaluate_all(&self, context: &EvaluationContext) -> Vec { match self.inner.current.load().as_ref() { @@ -244,26 +390,48 @@ impl Client { } } - /// Stops the background refresher. Also happens on drop. + /// Stops the background tasks. Also happens on drop. Unflushed events are + /// discarded — call [`Client::flush_events`] first if they matter. pub fn shutdown(&self) { - if let Some(handle) = &self.refresher { - handle.abort(); + if let Some(tasks) = &self.tasks { + for task in tasks.iter() { + task.abort(); + } } } } impl Drop for Client { fn drop(&mut self) { - // Only the last clone should stop the refresher; earlier ones going - // out of scope must not silently freeze everyone else's flags. - if let Some(handle) = &self.refresher - && Arc::strong_count(handle) == 1 + // Only the last clone should stop the tasks; earlier ones going out + // of scope must not silently freeze everyone else's flags. + if let Some(tasks) = &self.tasks + && Arc::strong_count(tasks) == 1 { - handle.abort(); + for task in tasks.iter() { + task.abort(); + } } } } +fn spawn_event_flusher(inner: Arc, interval: Duration) -> JoinHandle<()> { + tokio::spawn(async move { + loop { + tokio::time::sleep(interval).await; + // Failures keep their counts (see `flush_events`), so the only + // job here is to keep trying and to say what is happening. + if let Err(error) = inner.flush_events().await { + tracing::warn!( + %error, + transient = error.is_transient(), + "flagforge: event flush failed; counts kept for the next attempt" + ); + } + } + }) +} + fn spawn_refresher(inner: Arc, interval: Duration) -> JoinHandle<()> { tokio::spawn(async move { let mut backoff = interval; @@ -324,12 +492,14 @@ mod tests { #[test] fn trailing_slashes_do_not_produce_a_double_slash() { - // A zero interval skips the background task, so this needs no runtime. + // Zero intervals skip both background tasks, so this needs no runtime. let (_client, inner) = Client::builder("https://flags.example.com/", "ff_srv_x") .poll_interval(Duration::ZERO) + .event_flush_interval(Duration::ZERO) .build() .unwrap(); assert_eq!(inner.url, "https://flags.example.com/api/v1/snapshot"); + assert_eq!(inner.events_url, "https://flags.example.com/api/v1/events"); } #[tokio::test] diff --git a/crates/sdk/src/events.rs b/crates/sdk/src/events.rs new file mode 100644 index 0000000..9a32232 --- /dev/null +++ b/crates/sdk/src/events.rs @@ -0,0 +1,159 @@ +//! Experiment event accumulation. +//! +//! The SDK never sends individual events. Exposures and conversions land in an +//! in-process counter map — one cell per (experiment, variant, kind) — and the +//! flusher ships the counts. A service evaluating a flag a million times an +//! hour sends the same few dozen bytes as one evaluating it twice, which is +//! what lets the server keep pre-aggregated counters instead of an event log. + +use std::collections::HashMap; +use std::sync::Mutex; + +use serde::Serialize; + +/// Upper bound on distinct counter cells held between flushes. +/// +/// Cells are keyed by experiment × variant × kind, so a healthy process needs +/// a handful. Hitting this cap means experiment keys or variants are being +/// generated per request — a bug upstream — and the right failure mode is to +/// drop new cells and say so, not to grow without limit inside someone else's +/// service. +const MAX_CELLS: usize = 4_096; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum EventKind { + Exposure, + Conversion, +} + +/// One counter cell's identity. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct Cell { + pub experiment_key: String, + pub variant: String, + pub kind: EventKind, +} + +/// What `/api/v1/events` expects, borrowed straight out of a drained map. +#[derive(Debug, Serialize)] +pub(crate) struct EventPayload<'a> { + pub experiment_key: &'a str, + pub variant: &'a str, + pub kind: EventKind, + pub count: u32, +} + +/// The counter map, shared by every clone of the client. +#[derive(Debug, Default)] +pub(crate) struct Recorder { + state: Mutex, +} + +#[derive(Debug, Default)] +struct State { + counts: HashMap, + /// Increments refused by the cap since the last flush. + dropped: u64, +} + +impl Recorder { + pub fn record(&self, experiment_key: &str, variant: &str, kind: EventKind) { + let cell = + Cell { experiment_key: experiment_key.to_owned(), variant: variant.to_owned(), kind }; + let mut state = self.state.lock().expect("recorder mutex poisoned"); + + // Existing cells always increment; the cap only refuses *new* ones, so + // a well-behaved experiment keeps counting while a runaway one is + // contained. + if let Some(count) = state.counts.get_mut(&cell) { + *count += 1; + } else if state.counts.len() >= MAX_CELLS { + state.dropped += 1; + } else { + state.counts.insert(cell, 1); + } + } + + /// Takes everything recorded so far, leaving the map empty. + pub fn drain(&self) -> (HashMap, u64) { + let mut state = self.state.lock().expect("recorder mutex poisoned"); + (std::mem::take(&mut state.counts), std::mem::take(&mut state.dropped)) + } + + /// Puts a failed flush back so the next one retries it. Counts recorded in + /// the meantime are merged rather than overwritten. + pub fn restore(&self, counts: HashMap) { + let mut state = self.state.lock().expect("recorder mutex poisoned"); + for (cell, count) in counts { + if let Some(existing) = state.counts.get_mut(&cell) { + *existing += count; + } else if state.counts.len() >= MAX_CELLS { + state.dropped += count; + } else { + state.counts.insert(cell, count); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cell(experiment: &str, variant: &str) -> Cell { + Cell { + experiment_key: experiment.to_owned(), + variant: variant.to_owned(), + kind: EventKind::Exposure, + } + } + + #[test] + fn identical_events_pile_into_one_cell() { + let recorder = Recorder::default(); + for _ in 0..5 { + recorder.record("exp", "on", EventKind::Exposure); + } + recorder.record("exp", "on", EventKind::Conversion); + + let (counts, dropped) = recorder.drain(); + assert_eq!(counts.len(), 2); + assert_eq!(counts[&cell("exp", "on")], 5); + assert_eq!(dropped, 0); + let (rest, _) = recorder.drain(); + assert!(rest.is_empty(), "drain must leave the map empty"); + } + + #[test] + fn the_cap_refuses_new_cells_but_never_existing_ones() { + let recorder = Recorder::default(); + for i in 0..MAX_CELLS { + recorder.record(&format!("exp-{i}"), "on", EventKind::Exposure); + } + + // A brand-new cell is refused... + recorder.record("one-too-many", "on", EventKind::Exposure); + // ...but a cell that already exists keeps counting. + recorder.record("exp-0", "on", EventKind::Exposure); + + let (counts, dropped) = recorder.drain(); + assert_eq!(counts.len(), MAX_CELLS); + assert_eq!(dropped, 1); + assert_eq!(counts[&cell("exp-0", "on")], 2); + } + + #[test] + fn a_restored_flush_merges_with_what_arrived_meanwhile() { + let recorder = Recorder::default(); + recorder.record("exp", "on", EventKind::Exposure); + let (drained, _) = recorder.drain(); + + // The flush "failed"; traffic kept arriving. + recorder.record("exp", "on", EventKind::Exposure); + recorder.restore(drained); + + let (counts, _) = recorder.drain(); + assert_eq!(counts[&cell("exp", "on")], 2, "nothing may be lost or double-counted"); + } +} diff --git a/crates/sdk/src/lib.rs b/crates/sdk/src/lib.rs index 062404d..1fb3072 100644 --- a/crates/sdk/src/lib.rs +++ b/crates/sdk/src/lib.rs @@ -40,6 +40,7 @@ mod client; mod error; +mod events; pub use client::{Client, ClientBuilder, Snapshot}; pub use error::Error; diff --git a/crates/sdk/tests/agreement.rs b/crates/sdk/tests/agreement.rs index 6c92944..66d734b 100644 --- a/crates/sdk/tests/agreement.rs +++ b/crates/sdk/tests/agreement.rs @@ -165,6 +165,18 @@ impl Fixture { assert!(response.status().is_success(), "PUT {path} failed: {}", response.status()); } + async fn get(&self, path: &str) -> Value { + let response = self + .http + .get(format!("{}{path}", self.base)) + .header("authorization", format!("Bearer {}", self.token)) + .send() + .await + .unwrap(); + assert!(response.status().is_success(), "GET {path} failed: {}", response.status()); + response.json().await.unwrap() + } + /// Asks the server for a decision, the way a thin client would. async fn evaluate_remotely(&self, flag: &str, context: &EvaluationContext) -> bool { let response: Value = self @@ -303,6 +315,80 @@ async fn local_and_remote_agree_on_flags_gated_by_a_segment(pool: PgPool) { ); } +/// Exposures and conversions are attributed client-side, by the same +/// deterministic evaluation the flag check used. If that attribution drifted +/// from local assignment, an experiment would compare cohorts that do not +/// exist — so the counters the server ends up holding must equal a tally the +/// test keeps by hand. +#[sqlx::test(migrations = "../../migrations")] +async fn recorded_counters_agree_with_local_assignment(pool: PgPool) { + use std::collections::HashMap; + + let address = serve(pool).await; + let fixture = Fixture::create(address).await; + let experiments = "/api/v1/projects/checkout/environments/production/experiments"; + + fixture + .post( + experiments, + json!({ + "key": "cta", "name": "Checkout CTA", "flag_key": "checkout.v2", + "metric_key": "order.completed", "control_variant": "off", + }), + ) + .await; + fixture.post(&format!("{experiments}/cta/start"), json!({})).await; + + // Both background tasks off: the test drives the flush itself. + let client = Client::builder(&fixture.base, &fixture.sdk_key) + .poll_interval(Duration::ZERO) + .event_flush_interval(Duration::ZERO) + .connect() + .await + .unwrap(); + + let mut exposures: HashMap = HashMap::new(); + let mut conversions: HashMap = HashMap::new(); + for i in 0..200 { + let context = context(i); + let decision = + client.evaluate("checkout.v2", &context, flagforge_sdk::VariantValue::null()); + let variant = decision.variant.clone().expect("a configured flag resolves a variant"); + *exposures.entry(variant.clone()).or_default() += 1; + + // Every third user converts. + if i % 3 == 0 { + assert_eq!(client.track("order.completed", &context), 1); + *conversions.entry(variant).or_default() += 1; + } + } + + // A metric nobody measures is not an error — it counts toward nothing. + assert_eq!(client.track("unmeasured.metric", &context(0)), 0); + + client.flush_events().await.expect("the flush must reach the server"); + + let body = fixture.get(&format!("{experiments}/cta/results")).await; + let results = body["results"].as_array().expect("results array"); + assert_eq!(results.len(), 2, "{body}"); + for arm in results { + let variant = arm["variant"].as_str().unwrap(); + assert_eq!( + arm["exposures"].as_u64().unwrap(), + exposures.get(variant).copied().unwrap_or(0), + "exposures for `{variant}` disagree with local assignment: {body}" + ); + assert_eq!( + arm["conversions"].as_u64().unwrap(), + conversions.get(variant).copied().unwrap_or(0), + "conversions for `{variant}` disagree with local attribution: {body}" + ); + } + + // Flushing with nothing accumulated is a no-op, not a request. + client.flush_events().await.unwrap(); +} + #[sqlx::test(migrations = "../../migrations")] async fn the_sdk_sees_a_change_after_refreshing(pool: PgPool) { let address = serve(pool).await; diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs index 83bd9a6..2f682bf 100644 --- a/crates/storage/src/error.rs +++ b/crates/storage/src/error.rs @@ -32,6 +32,13 @@ pub enum StorageError { #[error("{entity} `{key}` is still referenced by {}", referenced_by.join(", "))] InUse { entity: &'static str, key: String, referenced_by: Vec }, + /// An operation that is only legal in some lifecycle state, attempted in + /// another — starting an already-stopped experiment, say. The message + /// names both states so the caller knows what to do rather than what + /// went wrong. + #[error("{entity} `{key}` is {actual}; this operation needs it {needed}")] + WrongState { entity: &'static str, key: String, actual: &'static str, needed: &'static str }, + /// A row we wrote no longer deserializes into the domain model. This means /// a schema/model skew, not a user error, so it is never a 4xx. #[error("stored {entity} is malformed: {source}")] diff --git a/crates/storage/src/experiments.rs b/crates/storage/src/experiments.rs new file mode 100644 index 0000000..7bd13f8 --- /dev/null +++ b/crates/storage/src/experiments.rs @@ -0,0 +1,457 @@ +//! A/B experiments and their pre-aggregated counters. + +use std::collections::HashMap; + +use chrono::{DateTime, DurationRound, TimeDelta, Utc}; +use flagforge_core::VariantCounts; +use sqlx::PgPool; +use uuid::Uuid; + +use crate::error::{FoundExt, Result, StorageError}; +use crate::models::{CounterDelta, CounterKind, Experiment, ExperimentState}; + +/// The columns every experiment read selects: the row itself plus the flag's +/// key and variants, which every consumer needs (see [`Experiment`]). +macro_rules! experiment_query { + ($condition:literal, $($arg:expr),* $(,)?) => { + sqlx::query!( + r#" + SELECT e.id, e.environment_id, e.flag_id, f.key AS flag_key, f.variants, + e.key, e.name, e.description, e.metric_key, e.control_variant, + e.state, e.started_at, e.stopped_at, e.version, e.created_at, e.updated_at + FROM experiments e + JOIN flags f ON f.id = e.flag_id + WHERE "# + $condition, + $($arg),* + ) + }; +} + +#[allow(clippy::too_many_arguments)] +pub async fn create_experiment( + pool: &PgPool, + environment_id: Uuid, + flag_id: Uuid, + key: &str, + name: &str, + description: Option<&str>, + metric_key: &str, + control_variant: &str, +) -> Result { + sqlx::query!( + r#" + INSERT INTO experiments + (environment_id, flag_id, key, name, description, metric_key, control_variant) + VALUES ($1, $2, $3, $4, $5, $6, $7) + "#, + environment_id, + flag_id, + key, + name, + description, + metric_key, + control_variant, + ) + .execute(pool) + .await + .map_err(|e| StorageError::from_write(e, "experiment", key))?; + + find_experiment(pool, environment_id, key).await +} + +pub async fn list_experiments(pool: &PgPool, environment_id: Uuid) -> Result> { + let rows = experiment_query!("e.environment_id = $1 ORDER BY e.key", environment_id) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|r| { + build( + r.id, + r.environment_id, + r.flag_id, + r.flag_key, + r.variants, + r.key, + r.name, + r.description, + r.metric_key, + r.control_variant, + r.state, + r.started_at, + r.stopped_at, + r.version, + r.created_at, + r.updated_at, + ) + }) + .collect() +} + +pub async fn find_experiment(pool: &PgPool, environment_id: Uuid, key: &str) -> Result { + let r = experiment_query!("e.environment_id = $1 AND e.key = $2", environment_id, key) + .fetch_optional(pool) + .await? + .or_not_found("experiment")?; + + build( + r.id, + r.environment_id, + r.flag_id, + r.flag_key, + r.variants, + r.key, + r.name, + r.description, + r.metric_key, + r.control_variant, + r.state, + r.started_at, + r.stopped_at, + r.version, + r.created_at, + r.updated_at, + ) +} + +/// Rewrites an experiment's presentation fields. +/// +/// The measurement fields — flag, metric, control — are deliberately not here: +/// changing what a running experiment measures would splice two different +/// questions into one answer. The API refuses those edits outside `draft`; a +/// draft's measurement fields are edited by deleting and recreating it, which +/// costs nothing because a draft has no counters yet. +pub async fn update_experiment( + pool: &PgPool, + environment_id: Uuid, + key: &str, + name: Option<&str>, + description: Option>, +) -> Result { + // "leave alone" vs "set to NULL", the same split update_segment uses. + let (clear_description, description_value) = match description { + None => (false, None), + Some(value) => (true, value), + }; + + let updated = sqlx::query!( + r#" + UPDATE experiments + SET name = COALESCE($3, name), + description = CASE WHEN $4 THEN $5 ELSE description END + WHERE environment_id = $1 AND key = $2 + "#, + environment_id, + key, + name, + clear_description, + description_value, + ) + .execute(pool) + .await + .map_err(|e| StorageError::from_write(e, "experiment", key))?; + + if updated.rows_affected() == 0 { + return Err(StorageError::not_found("experiment")); + } + find_experiment(pool, environment_id, key).await +} + +/// Moves a draft to `running`. One-way, enforced in the WHERE clause so two +/// racing starts cannot both succeed. +pub async fn start_experiment( + pool: &PgPool, + environment_id: Uuid, + key: &str, +) -> Result { + transition(pool, environment_id, key, ExperimentState::Draft, ExperimentState::Running).await +} + +/// Moves a running experiment to `stopped`, its terminal state. +pub async fn stop_experiment(pool: &PgPool, environment_id: Uuid, key: &str) -> Result { + transition(pool, environment_id, key, ExperimentState::Running, ExperimentState::Stopped).await +} + +async fn transition( + pool: &PgPool, + environment_id: Uuid, + key: &str, + from: ExperimentState, + to: ExperimentState, +) -> Result { + let updated = sqlx::query!( + r#" + UPDATE experiments + SET state = $4, + started_at = CASE WHEN $4 = 'running' THEN now() ELSE started_at END, + stopped_at = CASE WHEN $4 = 'stopped' THEN now() ELSE stopped_at END + WHERE environment_id = $1 AND key = $2 AND state = $3 + "#, + environment_id, + key, + from.as_str(), + to.as_str(), + ) + .execute(pool) + .await?; + + if updated.rows_affected() == 0 { + // Nothing moved: either the experiment is missing or it is in the + // wrong state. Reading it tells us which, and its actual state. + let current = find_experiment(pool, environment_id, key).await?; + return Err(StorageError::WrongState { + entity: "experiment", + key: key.to_owned(), + actual: current.state.as_str(), + needed: from.as_str(), + }); + } + find_experiment(pool, environment_id, key).await +} + +/// Deletes an experiment and, by cascade, its counters. +/// +/// A running experiment is refused: stopping is the explicit acknowledgement +/// that measurement is over, and deletion skipping it is usually a mis-click +/// about to destroy live results. +pub async fn delete_experiment(pool: &PgPool, environment_id: Uuid, key: &str) -> Result<()> { + let deleted = sqlx::query!( + r#"DELETE FROM experiments WHERE environment_id = $1 AND key = $2 AND state <> 'running'"#, + environment_id, + key, + ) + .execute(pool) + .await?; + + if deleted.rows_affected() == 0 { + let current = find_experiment(pool, environment_id, key).await?; + return Err(StorageError::WrongState { + entity: "experiment", + key: key.to_owned(), + actual: current.state.as_str(), + needed: "stopped (or still a draft)", + }); + } + Ok(()) +} + +/// Moves a running experiment's `started_at` into the past. +/// +/// Demo seeding only: seeded counters span the preceding week, and results +/// that predate their experiment's start would look like a data bug worth +/// investigating rather than like a demo. +pub async fn backdate_start( + pool: &PgPool, + environment_id: Uuid, + key: &str, + started_at: DateTime, +) -> Result<()> { + let updated = sqlx::query!( + r#" + UPDATE experiments SET started_at = $3 + WHERE environment_id = $1 AND key = $2 AND state = 'running' + "#, + environment_id, + key, + started_at, + ) + .execute(pool) + .await?; + + if updated.rows_affected() == 0 { + return Err(StorageError::not_found("experiment")); + } + Ok(()) +} + +/// Experiment keys that reference `flag_id`, qualified by environment so the +/// error message reading them locates each one. Used to refuse flag deletion. +pub async fn experiments_referencing_flag(pool: &PgPool, flag_id: Uuid) -> Result> { + let rows = sqlx::query!( + r#" + SELECT env.key AS environment_key, e.key + FROM experiments e + JOIN environments env ON env.id = e.environment_id + WHERE e.flag_id = $1 + ORDER BY env.key, e.key + "#, + flag_id, + ) + .fetch_all(pool) + .await?; + + Ok(rows.into_iter().map(|r| format!("{}/{}", r.environment_key, r.key)).collect()) +} + +/// Applies a batch of counter increments in one round trip. +/// +/// Returns how many increments were accepted. Deltas naming an experiment that +/// is unknown or not running are dropped rather than erroring: SDKs flush on a +/// timer, so a batch straddling an experiment's stop is routine traffic, not a +/// client bug to punish. +pub async fn ingest_counters( + pool: &PgPool, + environment_id: Uuid, + deltas: &[CounterDelta], +) -> Result { + // Collapse duplicates first: two deltas landing in the same counter cell + // would make `ON CONFLICT DO UPDATE` touch a row twice in one statement, + // which Postgres refuses outright. + type Cell = (String, String, CounterKind, DateTime); + let mut cells: HashMap = HashMap::new(); + for delta in deltas { + let hour = delta + .at + .duration_trunc(TimeDelta::hours(1)) + .map_err(|_| StorageError::Invalid { entity: "counter timestamp" })?; + let cell = (delta.experiment_key.clone(), delta.variant.clone(), delta.kind, hour); + *cells.entry(cell).or_default() += i64::from(delta.count); + } + + let mut keys = Vec::with_capacity(cells.len()); + let mut variants = Vec::with_capacity(cells.len()); + let mut kinds = Vec::with_capacity(cells.len()); + let mut hours = Vec::with_capacity(cells.len()); + let mut counts = Vec::with_capacity(cells.len()); + for ((key, variant, kind, hour), count) in cells { + keys.push(key); + variants.push(variant); + kinds.push(kind.as_str().to_owned()); + hours.push(hour); + counts.push(count); + } + + let written = sqlx::query!( + r#" + INSERT INTO experiment_counters (experiment_id, variant, kind, hour, count) + SELECT e.id, d.variant, d.kind, d.hour, d.count + FROM UNNEST($2::TEXT[], $3::TEXT[], $4::TEXT[], $5::TIMESTAMPTZ[], $6::BIGINT[]) + AS d(experiment_key, variant, kind, hour, count) + JOIN experiments e + ON e.environment_id = $1 AND e.key = d.experiment_key AND e.state = 'running' + ON CONFLICT (experiment_id, variant, kind, hour) + DO UPDATE SET count = experiment_counters.count + EXCLUDED.count + "#, + environment_id, + &keys, + &variants, + &kinds, + &hours, + &counts, + ) + .execute(pool) + .await?; + + Ok(written.rows_affected()) +} + +/// Total exposures and conversions per variant, summed across every hour. +/// +/// Variants appear only once they have a counter; the API layer fills in the +/// flag's other variants with zeroes so a results view shows every arm. +pub async fn experiment_counts(pool: &PgPool, experiment_id: Uuid) -> Result> { + let rows = sqlx::query!( + r#" + SELECT variant, kind, SUM(count)::BIGINT AS "total!" + FROM experiment_counters + WHERE experiment_id = $1 + GROUP BY variant, kind + ORDER BY variant + "#, + experiment_id, + ) + .fetch_all(pool) + .await?; + + let mut by_variant: Vec = Vec::new(); + for row in rows { + let entry = match by_variant.iter_mut().find(|c| c.variant == row.variant) { + Some(entry) => entry, + None => { + by_variant.push(VariantCounts { + variant: row.variant.clone(), + exposures: 0, + conversions: 0, + }); + by_variant.last_mut().expect("just pushed") + } + }; + let total = u64::try_from(row.total).unwrap_or(0); + match row.kind.as_str() { + "exposure" => entry.exposures = total, + _ => entry.conversions = total, + } + } + Ok(by_variant) +} + +/// The running experiments of one environment, for snapshot loading. +pub(crate) async fn load_running_experiments( + pool: &PgPool, + environment_id: Uuid, +) -> Result> { + let rows = sqlx::query!( + r#" + SELECT e.key, f.key AS flag_key, e.metric_key, e.control_variant, e.version + FROM experiments e + JOIN flags f ON f.id = e.flag_id + WHERE e.environment_id = $1 AND e.state = 'running' + ORDER BY e.key + "#, + environment_id, + ) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| flagforge_core::ExperimentSpec { + key: r.key, + flag_key: r.flag_key, + metric_key: r.metric_key, + control_variant: r.control_variant, + version: r.version, + }) + .collect()) +} + +#[allow(clippy::too_many_arguments)] +fn build( + id: Uuid, + environment_id: Uuid, + flag_id: Uuid, + flag_key: String, + variants: serde_json::Value, + key: String, + name: String, + description: Option, + metric_key: String, + control_variant: String, + state: String, + started_at: Option>, + stopped_at: Option>, + version: i64, + created_at: DateTime, + updated_at: DateTime, +) -> Result { + Ok(Experiment { + id, + environment_id, + flag_id, + flag_key, + variants: serde_json::from_value(variants) + .map_err(|e| StorageError::malformed("experiment flag variants", e))?, + key, + name, + description, + metric_key, + control_variant, + state: ExperimentState::parse(&state) + .ok_or(StorageError::Invalid { entity: "experiment state" })?, + started_at, + stopped_at, + version, + created_at, + updated_at, + }) +} diff --git a/crates/storage/src/flags.rs b/crates/storage/src/flags.rs index b773672..2b3bae3 100644 --- a/crates/storage/src/flags.rs +++ b/crates/storage/src/flags.rs @@ -196,7 +196,23 @@ pub async fn update_flag( }) } +/// Deletes a flag, refusing while any experiment still references it. +/// +/// An experiment's counters are only interpretable next to the flag whose +/// variants they tally, so the delete is refused with a message naming the +/// experiments — the same contract segment deletion has, though here a +/// RESTRICT foreign key backs it up at the database. pub async fn delete_flag(pool: &PgPool, project_id: Uuid, key: &str) -> Result<()> { + let flag = find_flag(pool, project_id, key).await?; + let referencing = crate::experiments::experiments_referencing_flag(pool, flag.id).await?; + if !referencing.is_empty() { + return Err(StorageError::InUse { + entity: "flag", + key: key.to_owned(), + referenced_by: referencing, + }); + } + let deleted = sqlx::query!(r#"DELETE FROM flags WHERE project_id = $1 AND key = $2"#, project_id, key) .execute(pool) diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 28a38ca..dc75afb 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -21,6 +21,7 @@ pub mod accounts; pub mod api_keys; pub mod audit; pub mod error; +pub mod experiments; pub mod flags; pub mod models; pub mod notify; diff --git a/crates/storage/src/models.rs b/crates/storage/src/models.rs index 4bd79c2..29940e9 100644 --- a/crates/storage/src/models.rs +++ b/crates/storage/src/models.rs @@ -179,6 +179,105 @@ impl Segment { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum ExperimentState { + /// Defined but not yet measuring; the only state where the measurement + /// fields may still be edited. + Draft, + /// Counting exposures and conversions; travels in the snapshot. + Running, + /// Finished. Terminal: reopening a measurement window would average two + /// populations into an answer about neither. + Stopped, +} + +impl ExperimentState { + pub fn as_str(self) -> &'static str { + match self { + Self::Draft => "draft", + Self::Running => "running", + Self::Stopped => "stopped", + } + } + + pub fn parse(raw: &str) -> Option { + match raw { + "draft" => Some(Self::Draft), + "running" => Some(Self::Running), + "stopped" => Some(Self::Stopped), + _ => None, + } + } +} + +/// An A/B experiment, as stored. +/// +/// `flag_key` and `variants` are denormalized from the flag at read time: every +/// consumer — validation, the dashboard, the results view — needs them, and +/// making each one re-fetch the flag would turn one join into four queries. +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct Experiment { + pub id: Uuid, + pub environment_id: Uuid, + pub flag_id: Uuid, + pub flag_key: String, + pub variants: Vec, + pub key: String, + pub name: String, + pub description: Option, + pub metric_key: String, + pub control_variant: String, + pub state: ExperimentState, + pub started_at: Option>, + pub stopped_at: Option>, + pub version: i64, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +impl Experiment { + /// The SDK-facing view, valid only while the experiment is running. + pub fn spec(&self) -> flagforge_core::ExperimentSpec { + flagforge_core::ExperimentSpec { + key: self.key.clone(), + flag_key: self.flag_key.clone(), + metric_key: self.metric_key.clone(), + control_variant: self.control_variant.clone(), + version: self.version, + } + } +} + +/// What one kind of counter counts. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum CounterKind { + /// The flag was evaluated for a context while the experiment ran. + Exposure, + /// A tracked metric event attributed to a variant. + Conversion, +} + +impl CounterKind { + pub fn as_str(self) -> &'static str { + match self { + Self::Exposure => "exposure", + Self::Conversion => "conversion", + } + } +} + +/// One increment on its way into `experiment_counters`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CounterDelta { + pub experiment_key: String, + pub variant: String, + pub kind: CounterKind, + pub at: DateTime, + pub count: u32, +} + #[derive(Debug, Clone, Serialize, ToSchema)] pub struct ApiKey { pub id: Uuid, diff --git a/crates/storage/src/snapshot.rs b/crates/storage/src/snapshot.rs index 7c964c9..3f92e0a 100644 --- a/crates/storage/src/snapshot.rs +++ b/crates/storage/src/snapshot.rs @@ -6,14 +6,16 @@ use sqlx::PgPool; use uuid::Uuid; use crate::error::{FoundExt, Result, StorageError}; +use crate::experiments::load_running_experiments; use crate::flags::load_environment_flags; use crate::segments::load_environment_segments; /// Loads everything one environment needs to answer evaluations. /// -/// Three queries: the environment (for its salt), every configured flag, and -/// every segment their rules might reference. The result is self-contained, so -/// once it is in memory the evaluation path never touches Postgres again. +/// Four queries: the environment (for its salt), every configured flag, every +/// segment their rules might reference, and the running experiments SDKs +/// should be recording. The result is self-contained, so once it is in memory +/// the evaluation path never touches Postgres again. pub async fn load(pool: &PgPool, environment_id: Uuid) -> Result { let env = sqlx::query!(r#"SELECT key, salt FROM environments WHERE id = $1"#, environment_id,) .fetch_optional(pool) @@ -39,8 +41,10 @@ pub async fn load(pool: &PgPool, environment_id: Uuid) -> Result, } + /// An A/B experiment, as the management API returns it. + #[derive(Debug, Clone, PartialEq, Deserialize)] + pub struct Experiment { + pub id: String, + pub key: String, + pub name: String, + pub description: Option, + pub flag_key: String, + /// The flag's variants, denormalized so the results view can render + /// every arm without a second request. + pub variants: Vec, + pub metric_key: String, + pub control_variant: String, + /// `draft`, `running` or `stopped`. + pub state: String, + pub started_at: Option, + pub stopped_at: Option, + pub version: i64, + } + + /// An experiment next to its judged results. + #[derive(Debug, Clone, PartialEq, Deserialize)] + pub struct ExperimentResults { + pub experiment: Experiment, + pub results: Vec, + } + #[derive(Debug, Clone, PartialEq, Deserialize)] pub struct ApiKey { pub id: String, @@ -323,6 +350,17 @@ pub struct SegmentBody { pub expected_version: i64, } +#[derive(Serialize)] +pub struct NewExperiment<'a> { + pub key: &'a str, + pub name: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + pub description: Option<&'a str>, + pub flag_key: &'a str, + pub metric_key: &'a str, + pub control_variant: &'a str, +} + #[derive(Serialize)] pub struct NewKey<'a> { pub name: &'a str, @@ -596,6 +634,84 @@ pub async fn delete_segment( .await } +pub async fn list_experiments( + token: &str, + project: &str, + environment: &str, +) -> Result, ApiError> { + send( + Method::Get, + &format!("/projects/{project}/environments/{environment}/experiments"), + Some(token), + None, + ) + .await +} + +pub async fn create_experiment( + token: &str, + project: &str, + environment: &str, + body: NewExperiment<'_>, +) -> Result { + send( + Method::Post, + &format!("/projects/{project}/environments/{environment}/experiments"), + Some(token), + json(body), + ) + .await +} + +/// `action` is `start` or `stop` — the two lifecycle transitions. +pub async fn transition_experiment( + token: &str, + project: &str, + environment: &str, + experiment: &str, + action: &str, +) -> Result { + send( + Method::Post, + &format!( + "/projects/{project}/environments/{environment}/experiments/{experiment}/{action}" + ), + Some(token), + Some(Value::Null), + ) + .await +} + +pub async fn experiment_results( + token: &str, + project: &str, + environment: &str, + experiment: &str, +) -> Result { + send( + Method::Get, + &format!("/projects/{project}/environments/{environment}/experiments/{experiment}/results"), + Some(token), + None, + ) + .await +} + +pub async fn delete_experiment( + token: &str, + project: &str, + environment: &str, + experiment: &str, +) -> Result<(), ApiError> { + send( + Method::Delete, + &format!("/projects/{project}/environments/{environment}/experiments/{experiment}"), + Some(token), + None, + ) + .await +} + pub async fn list_keys( token: &str, project: &str, diff --git a/crates/web/src/app.rs b/crates/web/src/app.rs index 48a69ac..cd45bf8 100644 --- a/crates/web/src/app.rs +++ b/crates/web/src/app.rs @@ -32,6 +32,7 @@ pub fn App() -> impl IntoView { + diff --git a/crates/web/src/components.rs b/crates/web/src/components.rs index b878705..3f2ff34 100644 --- a/crates/web/src/components.rs +++ b/crates/web/src/components.rs @@ -44,6 +44,9 @@ fn path_for(name: &str) -> &'static str { r#""# } "moon" => r#""#, + "flask" => { + r#""# + } "plus" => r#""#, "trash" => { r#""# diff --git a/crates/web/src/pages/experiments.rs b/crates/web/src/pages/experiments.rs new file mode 100644 index 0000000..79f6cb0 --- /dev/null +++ b/crates/web/src/pages/experiments.rs @@ -0,0 +1,858 @@ +//! A/B experiments for one environment. +//! +//! Organised like the segments page — pick an environment, then work inside +//! it — because an experiment *is* a measurement of one environment's traffic. +//! The list shows lifecycle at a glance; opening an experiment shows its +//! judged results: rates, intervals and the verdict against the control. + +use flagforge_core::VariantResult; +use leptos::prelude::*; +use leptos_router::components::A; +use leptos_router::hooks::use_params_map; + +use crate::api::{ + self, Load, NewExperiment, + models::{ConfiguredFlag, Environment, Experiment, ExperimentResults}, +}; +use crate::app::{PageHeader, fixed}; +use crate::components::{ConfirmButton, Empty, Failure, Icon, Modal, SkeletonRows, defer}; +use crate::pages::projects_slug; +use crate::session::use_session; +use crate::toast::use_toaster; + +#[component] +pub fn Experiments() -> impl IntoView { + let session = use_session(); + let toaster = use_toaster(); + let params = use_params_map(); + + let project_key = move || params.read().get("project").unwrap_or_default(); + + let environments = RwSignal::new(Load::>::Loading); + let selected_env = RwSignal::new(Option::::None); + let experiments = RwSignal::new(Load::>::Loading); + let opened = RwSignal::new(Option::>::None); + let opened_key = RwSignal::new(Option::::None); + let creating = RwSignal::new(false); + + let load_environments = move || { + let (Some(token), project) = (session.token_untracked(), project_key()) else { return }; + leptos::task::spawn_local(async move { + let list = api::list_environments(&token, &project).await; + if let Ok(list) = &list { + selected_env.set( + list.iter() + .find(|e| e.is_production) + .or_else(|| list.first()) + .map(|e| e.key.clone()), + ); + } + environments.set(list.into()); + }); + }; + + let load_experiments = move || { + let (Some(token), project) = (session.token_untracked(), project_key()) else { return }; + let Some(environment) = selected_env.get_untracked() else { return }; + experiments.set(Load::Loading); + leptos::task::spawn_local(async move { + experiments.set(api::list_experiments(&token, &project, &environment).await.into()); + }); + }; + + let open = move |key: String| { + let (Some(token), project) = (session.token_untracked(), project_key()) else { return }; + let Some(environment) = selected_env.get_untracked() else { return }; + opened_key.set(Some(key.clone())); + opened.set(Some(Load::Loading)); + leptos::task::spawn_local(async move { + opened.set(Some( + api::experiment_results(&token, &project, &environment, &key).await.into(), + )); + }); + }; + + Effect::new(move |_| { + let _ = project_key(); + load_environments(); + }); + + Effect::new(move |_| { + let _ = selected_env.get(); + opened.set(None); + opened_key.set(None); + load_experiments(); + }); + + let can_write = move || session.identity().is_some_and(|me| me.user.can_write()); + + // Start and stop share everything but the verb and the toast. + let transition = move |key: String, action: &'static str| { + let (Some(token), project) = (session.token_untracked(), project_key()) else { return }; + let Some(environment) = selected_env.get_untracked() else { return }; + leptos::task::spawn_local(async move { + match api::transition_experiment(&token, &project, &environment, &key, action).await { + Ok(_) => { + toaster.success(if action == "start" { + "Experiment running — SDKs will start recording" + } else { + "Experiment stopped — the results are final" + }); + load_experiments(); + if opened_key.get_untracked().as_deref() == Some(key.as_str()) { + open(key); + } + } + Err(error) => toaster.error("Could not change the experiment", &error), + } + }); + }; + + let delete = move |key: String| { + let (Some(token), project) = (session.token_untracked(), project_key()) else { return }; + let Some(environment) = selected_env.get_untracked() else { return }; + leptos::task::spawn_local(async move { + match api::delete_experiment(&token, &project, &environment, &key).await { + Ok(()) => { + toaster.success("Experiment deleted"); + opened.set(None); + opened_key.set(None); + load_experiments(); + } + Err(error) => toaster.error("Could not delete the experiment", &error), + } + }); + }; + + view! { + + + + "Back" + + + + + + +
+ {move || match environments.get() { + Load::Ready(list) => { + view! { +
+ {list + .into_iter() + .map(|environment| { + let key = environment.key.clone(); + let active = { + let key = key.clone(); + Signal::derive(move || { + selected_env.get().as_deref() == Some(key.as_str()) + }) + }; + view! { + + } + }) + .collect_view()} +
+ } + .into_any() + } + _ => view! {
}.into_any(), + }} + + {move || match experiments.get() { + Load::Loading => { + view! {
}.into_any() + } + Load::Failed(error) => { + view! { +
+ +
+ } + .into_any() + } + Load::Ready(list) if list.is_empty() => { + view! { +
+ +
+ } + .into_any() + } + Load::Ready(list) => { + view! { +
+
+ + + + + + + + + + + {list + .into_iter() + .map(|experiment| { + let key = experiment.key.clone(); + let open_key = key.clone(); + let start_key = key.clone(); + let stop_key = key.clone(); + let state = experiment.state.clone(); + let is_draft = state == "draft"; + let is_running = state == "running"; + // Callbacks are Copy, so the Show children stay Fn. + let start_now = Callback::new(move |_: ()| { + transition(start_key.clone(), "start") + }); + let stop_now = Callback::new(move |_: ()| { + transition(stop_key.clone(), "stop") + }); + view! { + + + + + + + } + }) + .collect_view()} + +
"Key""Measures""State"
+ {experiment.key.clone()} +
{experiment.name.clone()}
+ 		+ {experiment.flag_key.clone()} + " → " + {experiment.metric_key.clone()} + + + + + + + + +
+
+
+ } + .into_any() + } + }} + + {move || { + opened + .get() + .map(|load| match load { + Load::Loading => { + view! {
}.into_any() + } + Load::Failed(error) => { + view! { +
+ +
+ } + .into_any() + } + Load::Ready(loaded) => { + let key = loaded.experiment.key.clone(); + let delete_this = Callback::new(move |_| delete(key.clone())); + view! { + + } + .into_any() + } + }) + }} +
+ + + + + } +} + +#[component] +fn StateBadge(state: String) -> impl IntoView { + // The dot plus the word: state is never colour alone. + match state.as_str() { + "running" => view! { + + + "Running" + + } + .into_any(), + "stopped" => view! { "Stopped" }.into_any(), + _ => view! { "Draft" }.into_any(), + } +} + +/// The results view: one meter per arm on a shared scale, numbers beside it, +/// and the z-test's verdict in words. +#[component] +fn ResultsPanel( + loaded: ExperimentResults, + can_write: Signal, + #[prop(into)] on_close: Callback<()>, + #[prop(into)] on_delete: Callback<()>, +) -> impl IntoView { + let experiment = loaded.experiment; + let running = experiment.state == "running"; + let control = experiment.control_variant.clone(); + + // One scale for every arm, snapped up to the next 5 % so the bars have + // headroom and the axis note reads as a round number. + let top = loaded + .results + .iter() + .filter_map(|r| r.interval.map(|ci| ci.high).or(r.rate)) + .fold(0.0_f64, f64::max); + let scale = ((top / 0.05).ceil() * 0.05).max(0.05); + + let no_data = loaded.results.iter().all(|r| r.exposures == 0 && r.conversions == 0); + let has_excess = loaded.results.iter().any(|r| r.exposures > 0 && r.conversions > r.exposures); + + // Colour is keyed to the arm's position in the flag definition, matching + // the distribution bar on the flag page, so an arm keeps its colour + // between the two views. + let position_of = { + let variants: Vec = experiment.variants.iter().map(|v| v.key.clone()).collect(); + move |variant: &str| variants.iter().position(|v| v == variant).unwrap_or(0) % 4 + }; + + view! { +
+
+

+ {experiment.name.clone()} + " " + +

+
+ +
+ +
+
+ "Flag" + {experiment.flag_key.clone()} + "Metric" + {experiment.metric_key.clone()} + "Control" + {control.clone()} + {experiment + .started_at + .clone() + .map(|at| { + view! { + "Started" + {at.chars().take(10).collect::()} + } + })} + {experiment + .stopped_at + .clone() + .map(|at| { + view! { + "Stopped" + {at.chars().take(10).collect::()} + } + })} +
+ + +
+
+ +
+ + "No events yet. Exposures arrive when SDKs evaluate the flag; conversions when they call " + "track(\"" {experiment.metric_key.clone()} "\")" + "." + +
+ + + +
+
+ +
+ + "An arm has more conversions than exposures — the metric is being tracked for contexts that never evaluated the flag. Rates are clamped, raw counts shown." + +
+ + +
+ + + + + + + + + + + + {loaded + .results + .iter() + .map(|arm| { + let is_control = arm.variant == control; + let position = position_of(&arm.variant); + view! { + + } + }) + .collect_view()} + +
"Arm""Exposures""Conversions" + "Conversion rate" + + {format!(" (scale 0–{:.0} %)", scale * 100.0)} + + "Against control"
+
+
+ +
+ + + "Running experiments cannot be deleted — stop it first." + + } + } + > + + + +
+
+
+ } +} + +#[component] +fn ArmRow(arm: VariantResult, is_control: bool, position: usize, scale: f64) -> impl IntoView { + let rate_text = arm.rate.map(|rate| format!("{:.1} %", rate * 100.0)); + let ci_text = + arm.interval.map(|ci| format!("95 % CI {:.1}–{:.1} %", ci.low * 100.0, ci.high * 100.0)); + let meter_label = match (&rate_text, &ci_text) { + (Some(rate), Some(ci)) => format!("{}: {rate}, {ci}", arm.variant), + _ => format!("{}: no exposures yet", arm.variant), + }; + + let fill_width = arm.rate.map(|rate| format!("width:{:.2}%", (rate / scale) * 100.0)); + let ci_style = arm.interval.map(|ci| { + format!( + "left:{:.2}%;width:{:.2}%", + (ci.low / scale) * 100.0, + ((ci.high - ci.low) / scale).max(0.0) * 100.0, + ) + }); + + view! { + + + {arm.variant.clone()} + + "control" + + + {arm.exposures} + {arm.conversions} + + +
+ {match (rate_text, ci_text) { + (Some(rate), Some(ci)) => format!("{rate} · {ci}"), + _ => "No exposures yet".to_owned(), + }} +
+ + {verdict(&arm, is_control)} + + } +} + +/// The comparison cell, in words. The badge never carries the verdict alone. +fn verdict(arm: &VariantResult, is_control: bool) -> impl IntoView + use<> { + if is_control { + return view! { "— baseline" }.into_any(); + } + match arm.vs_control { + Some(comparison) => { + let lift = format!("{:+.1} pts", comparison.lift * 100.0); + let p = if comparison.p_value < 0.001 { + "p < 0.001".to_owned() + } else { + format!("p = {:.3}", comparison.p_value) + }; + let badge = if comparison.significant { + view! { + + + "Significant" + + } + .into_any() + } else { + view! { "Not significant" }.into_any() + }; + view! { +
+ {badge} + + {format!("{lift} · {p}")} + +
+ } + .into_any() + } + None => view! { "Awaiting data" }.into_any(), + } +} + +#[component] +fn CreateExperiment( + project: Signal, + environment: Signal, + #[prop(into)] on_close: Callback<()>, + #[prop(into)] on_created: Callback, +) -> impl IntoView { + let session = use_session(); + let toaster = use_toaster(); + + let flags = RwSignal::new(Load::>::Loading); + let name = RwSignal::new(String::new()); + let key = RwSignal::new(String::new()); + let key_touched = RwSignal::new(false); + let flag_key = RwSignal::new(String::new()); + let metric_key = RwSignal::new(String::new()); + let control = RwSignal::new(String::new()); + let busy = RwSignal::new(false); + + // The flag picker needs the environment's flags; their variants feed the + // control picker, and the off variant is the natural default control — + // it is what traffic outside the experiment would have seen. + Effect::new(move |_| { + let (Some(token), project, environment) = + (session.token_untracked(), project.get(), environment.get()) + else { + return; + }; + leptos::task::spawn_local(async move { + let list = api::list_configured(&token, &project, &environment).await; + if let Ok(list) = &list + && let Some(first) = list.iter().find(|c| !c.flag.archived) + { + flag_key.set(first.flag.key.clone()); + control.set(first.config.off_variant.clone()); + } + flags.set(list.into()); + }); + }); + + let variants_of_selected = Signal::derive(move || match flags.get() { + Load::Ready(list) => list + .iter() + .find(|c| c.flag.key == flag_key.get()) + .map(|c| c.flag.variants.clone()) + .unwrap_or_default(), + _ => Vec::new(), + }); + + let submit = move |event: leptos::ev::SubmitEvent| { + event.prevent_default(); + if busy.get_untracked() { + return; + } + busy.set(true); + + let Some(token) = session.token_untracked() else { return }; + let (project, environment) = (project.get_untracked(), environment.get_untracked()); + let (name_value, key_value, flag_value, metric_value, control_value) = ( + name.get_untracked(), + key.get_untracked(), + flag_key.get_untracked(), + metric_key.get_untracked(), + control.get_untracked(), + ); + + leptos::task::spawn_local(async move { + let result = api::create_experiment( + &token, + &project, + &environment, + NewExperiment { + key: &key_value, + name: &name_value, + description: None, + flag_key: &flag_value, + metric_key: &metric_value, + control_variant: &control_value, + }, + ) + .await; + + busy.set(false); + match result { + Ok(created) => on_created.run(created.key), + Err(error) => toaster.error("Could not create the experiment", &error), + } + }); + }; + + view! { + +
+
+
+ + +
+ +
+ + +
+ +
+ + + + "Its variants become the arms. Fixed once created." + +
+ +
+ + + + "What your service passes to " "track()" + " when the thing you care about happens." + +
+ +
+ + + + "The baseline every other arm is judged against — usually the behaviour users had before." + +
+ +
+
+ +
+ + "Created as a draft. Nothing is measured until you press Start, and a stopped experiment cannot restart." + +
+
+ +
+
+ + +
+
+ + } +} diff --git a/crates/web/src/pages/mod.rs b/crates/web/src/pages/mod.rs index a0210d2..0408c7e 100644 --- a/crates/web/src/pages/mod.rs +++ b/crates/web/src/pages/mod.rs @@ -1,6 +1,7 @@ //! Route views. mod audit; +mod experiments; pub(crate) mod flag; mod keys; mod login; @@ -9,6 +10,7 @@ mod projects; mod segments; pub use audit::Audit; +pub use experiments::Experiments; pub use flag::FlagDetail; pub use keys::Keys; pub use login::Login; diff --git a/crates/web/src/pages/project.rs b/crates/web/src/pages/project.rs index 181f9f6..0f052c1 100644 --- a/crates/web/src/pages/project.rs +++ b/crates/web/src/pages/project.rs @@ -172,6 +172,13 @@ pub fn ProjectDetail() -> impl IntoView { "Segments" + + + "Experiments" + running -> stopped, one way. A stopped experiment cannot + -- restart: its counters describe one measurement window, and reopening + -- the window would average two populations into an answer about neither. + state TEXT NOT NULL DEFAULT 'draft' + CHECK (state IN ('draft', 'running', 'stopped')), + started_at TIMESTAMPTZ, + stopped_at TIMESTAMPTZ, + -- Bumped by trigger on every write. Running experiments fold into the + -- snapshot version, so starting or stopping one is visible to SDKs. + version BIGINT NOT NULL DEFAULT 1, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX experiments_environment_key_key ON experiments (environment_id, key); + +-- The snapshot loader wants only the running ones, the same shape of partial +-- index flags_project_active_idx uses for unarchived flags. +CREATE INDEX experiments_running_idx ON experiments (environment_id) WHERE state = 'running'; + +-- Reusing the segments trigger contract: monotonic per row, set by the +-- database so racing writers cannot publish the same version twice. +CREATE TRIGGER experiments_touch + BEFORE INSERT OR UPDATE ON experiments + FOR EACH ROW +EXECUTE FUNCTION flagforge_touch_segment(); + +-- Experiments live on environment_id, so the existing notify function reads +-- the right column. +CREATE TRIGGER experiments_notify + AFTER INSERT OR UPDATE OR DELETE ON experiments + FOR EACH ROW +EXECUTE FUNCTION flagforge_notify_change(); + +-- Pre-aggregated tallies: one row per (variant, kind, hour), incremented in +-- place. The table is bounded by hours elapsed times variants, not by +-- traffic — the write path a raw event log would need is exactly the +-- infrastructure this schema exists to avoid. The price, accepted knowingly, +-- is that individual events are gone: there is nothing to re-analyse later. +CREATE TABLE experiment_counters ( + experiment_id UUID NOT NULL REFERENCES experiments (id) ON DELETE CASCADE, + variant TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('exposure', 'conversion')), + hour TIMESTAMPTZ NOT NULL, + count BIGINT NOT NULL CHECK (count >= 0), + PRIMARY KEY (experiment_id, variant, kind, hour) +); + +-- No touch or notify triggers here, deliberately. Counter increments arrive +-- continuously while an experiment runs; a notify per increment would turn +-- the ingest path into a snapshot-cache invalidation storm, and nothing an +-- SDK evaluates depends on a tally.