diff --git a/docs/README.md b/docs/README.md index 3a4711a1..ae6f97ce 100644 --- a/docs/README.md +++ b/docs/README.md @@ -33,6 +33,7 @@ Solve a specific problem: Look up exact behavior: +- [Consequence delta](reference/consequence-delta.md) - [Failure reasons](reference/failure-reasons.md) - [MCP tool reference](reference/mcp-tools.md) - [MCP tool profiles](reference/mcp-tool-profiles.md) @@ -82,6 +83,7 @@ Understand how Atlas works and why: - [Board renderer performance](explanation/board-renderer-performance.md) - [Branching](explanation/branching.md) - [Changesets](explanation/changesets.md) +- [Consequence deltas](explanation/consequence-delta.md) - [Entity protection](explanation/entity-protection.md) - [Reliability overlays](explanation/reliability-overlays.md) - [Single-writer fencing](explanation/single-writer.md) diff --git a/docs/explanation/consequence-delta.md b/docs/explanation/consequence-delta.md new file mode 100644 index 00000000..8be5fb02 --- /dev/null +++ b/docs/explanation/consequence-delta.md @@ -0,0 +1,151 @@ +# Consequence deltas + +A dry run answers "would this write be accepted?". A reviewer usually also +wants to know "what would it do to the model's health?": which findings it +creates, which it clears, and which it leaves alone. A consequence delta +answers that second question by comparing the findings of the world as it +is with the findings of the world the change would produce. + +The delta is advisory. It never refuses a write and never changes the +status a dry run would otherwise return. It rides the dry-run responses of +`PutEntity`, `BatchMutate` and `MergeBranch`, and the matching MCP tools. +Field-level detail lives in the +[consequence delta reference](../reference/consequence-delta.md). + +## Full-scope comparison is the reference + +The server scans the whole visible world twice, once before and once after +the change, with the same suites, and compares the two finding sets. It does +not try to work out which findings a change could reach. That kind of +narrowing is attractive, since most changes touch a small corner of the +graph, but a narrowing that misses a reverse reference silently drops a +real consequence. Full scope is the behaviour any later optimization must +reproduce exactly, and it is the only behaviour shipped today. + +Running every suite over the whole world costs time on every dry run, twice, +and the per-model suite grows faster than the world does because each model +pass reads the whole snapshot. The server records that cost on the +`trogon_atlas.consequence_delta.duration` histogram, so an operator can see +when a narrower comparison becomes worth building. + +Because each scan is a flat pass over a snapshot rather than a walk along +references, reference cycles cannot make it loop. + +## What makes two findings the same + +Prose changes between releases and must not turn one finding into two, so a +finding's identity is built only from structured data: + +- the suite that produced it, +- the entity whose pass produced it, for suites that run once per entity, +- the rule code, +- the exact subject entity, including its version, +- the subject field, +- every remaining structured field of the issue. + +The producing entity matters because a per-model check reports its finding +on the member it is about, not on the model. Without it, two models that +both point at the same missing event would collapse into one finding, and +deleting one of them would look like a fix. + +Severity is deliberately outside identity, so a rule that is promoted from +warning to error shows up as one finding whose severity changed rather than +one resolved and one introduced. The rule title, category and default +severity are copies of catalogue data and are excluded too. So is each +evidence entry's `detail`, which is prose like the message. + +"Every remaining structured field" is computed from the issue message itself +rather than from a fixed list. When `ValidationIssue` gains structured +evidence, that evidence joins identity without any change to the delta, so +two findings of one rule on one subject that differ only in evidence stay +distinct. + +## How findings are classified + +Findings are grouped by identity on each side and then matched: + +- Occurrences with equal identity and equal severity cancel out and are + counted as unchanged. +- Leftover occurrences with equal identity but different severity are a + severity change. +- A finding that appears on a newer version of its subject or producing + entity, when the same finding existed on an older version, is carried to + the successor rather than introduced. The entity versions its evidence + pins count the same way, so a slice that cites itself still carries its + finding to its next version, and evidence that moved to an older version + is not a successor. +- A finding whose subject or producing entity existed before and is gone + after left with that entity, which is not the same as being fixed. +- Anything else that appears is introduced, and anything else that + disappears is resolved. + +Repeated occurrences of one identity are counted, not collapsed, so a +change that adds a second copy of an existing finding is reported. + +## Unknown is not resolved + +A finding can only be called resolved if the after-scan actually looked +for it. If the suite that produced it did not complete after the change +(skipped, partial, errored, or simply not reported), its absence proves +nothing, and the delta reports it as unknown with the reason the suite +gave. The exception is a finding whose subject or origin no longer exists: +the change removed what the finding was about, so the delta reports it as +subject removed even when the after-scan's suite did not complete. The same holds the other way: a finding only counts as introduced if +the before-scan's suite completed, otherwise it might have been there all +along. + +A truncated snapshot marks every compared suite partial, so a delta over a +truncated world classifies nothing as introduced or resolved. The delta's +`complete` flag is true only when every compared suite completed on both +sides, neither snapshot was truncated, and no applicable suite was left +out. + +Both scans see only what the caller may read. An entity hidden from the +caller contributes no findings to either side, so the delta says nothing +about it rather than calling its findings resolved. A finding whose subject +was deleted is reported as removed with its subject, never as resolved. + +## Suites the delta does not compare + +Two parts of validation read the store directly rather than a candidate +world: specification type checks and data-product event checks. The delta +cannot evaluate them against a world that does not exist yet. When the +world holds entities those checks apply to, the delta names them as skipped +with `delta_not_supported` and reports itself incomplete. When it does not, +they are skipped as not applicable and do not affect completeness. + +Data-flow inference and information-completeness checks are scoped +analyses, optionally backed by a language model, so two runs over one world +need not agree. They are listed as skipped with `analysis_not_compared` and +do not affect completeness, because they are not validation suites. + +## Binding to the worlds compared + +Each delta carries the snapshot id of both worlds, computed the same way as +[`GetSnapshotId`](snapshot-identity.md). The before id names the state the +preview started from; the after id names the content that would result. A +delta whose before id no longer matches the live snapshot describes a +world that has moved on. When either world holds an entity that cannot be +rendered canonically, it has no snapshot id; the delta still reports its +findings but leaves both ids empty and calls itself incomplete, because a +comparison that cannot name its worlds cannot be checked for staleness. + +The ids cover only the two compared worlds. When the snapshot is truncated, +architecture validation also loads the targets a resource references from +the store, and a target outside both worlds can shape a finding without +being hashed into either id. Such a delta already calls itself incomplete, +so a matching before id does not prove those findings current. Binding that +context into the ids would change the id format and is a separate change. + +The comparison runs after the dry run releases the mutation lock. The lock +only covers the same admission checks a real write would make; the scans +read worlds already held in memory, so a slow delta never delays other +writers. The price is that a concurrent write can land while the delta is +being computed, which is exactly what the before snapshot id lets a reader +detect. + +## Default chosen for review + +The delta compares full scope only and reports applicable suites it cannot +evaluate as gaps rather than approximating them. Both choices trade cost and +coverage for never reporting a resolution that was not proven. diff --git a/docs/explanation/validation-coverage.md b/docs/explanation/validation-coverage.md index 5a3d381f..62b85ea3 100644 --- a/docs/explanation/validation-coverage.md +++ b/docs/explanation/validation-coverage.md @@ -152,9 +152,11 @@ subset). Coverage work makes that answer observable either way. Truncation, depth, and visibility flags must use the same scope and snapshot identity vocabulary as coverage, so a truncated empty issue list -cannot look complete. Consequence deltas then compare findings only within -declared comparable scopes; an after-scan that omitted a suite or truncated -the snapshot cannot claim a finding was resolved. +cannot look complete. [Consequence deltas](consequence-delta.md) build on +the same coverage: each side of the comparison carries its own coverage, +and a finding missing from an after-scan whose suite did not complete is +reported as unknown, never as resolved, unless its subject or origin was +removed, in which case it is reported as subject removed. ## Where the code lives diff --git a/docs/reference/consequence-delta.md b/docs/reference/consequence-delta.md new file mode 100644 index 00000000..52d1f78d --- /dev/null +++ b/docs/reference/consequence-delta.md @@ -0,0 +1,99 @@ +# Consequence delta + +Wire contract for +[`ConsequenceDelta`](../../proto/trogonatlas/api/eventmodel/v1alpha1/validation.proto). +For the reasoning behind it, see the +[consequence delta explanation](../explanation/consequence-delta.md). + +## Where it appears + +| Response | Field | Set when | +| --- | --- | --- | +| `PutEntityResponse` | `consequence_delta` | `validate_only` was set and the write would be admitted | +| `BatchMutateResponse` | `consequence_delta` | `validate_only` was set and the status is `STATUS_VALIDATED` | +| `MergeBranchResponse` | `consequence_delta` | `dry_run` was set and validation was reached (`STATUS_VALIDATED` or `STATUS_INVALID`) | + +A real write never carries a delta. An absent delta means none was +computed, never that nothing changed. Failing to compute a delta never +fails the dry run it rides on. + +The MCP tools `put_entity`, `put_entity_json`, `batch_mutate`, +`batch_mutate_json`, `apply_manifests` and `merge_branch`, and the +`trogon-atlas apply --dry-run` and `trogon-atlas branch merge --dry-run` +JSON output, add a `consequence_delta` object with the same fields. + +`apply_manifests` and `trogon-atlas apply --dry-run` send no batch when +every manifest is unchanged, so they carry no delta in that case. + +## `ConsequenceDelta` + +| Field | Meaning | +| --- | --- | +| `changes` | Findings whose standing moved, in a stable order | +| `unchanged_count` | Occurrences present on both sides with the same severity | +| `before_coverage` | Coverage of the scan of the world before the change | +| `after_coverage` | Coverage of the scan of the world after the change | +| `complete` | True only when every compared suite completed on both sides, no snapshot was truncated, no applicable suite was left out, and both snapshot ids are set | +| `before_snapshot_id` | Snapshot id of the world the preview started from; empty when that world holds an entity without content identity | +| `after_snapshot_id` | Snapshot id of the world the change would produce; empty under the same condition | + +## `FindingChange` + +| Field | Meaning | +| --- | --- | +| `kind` | See the kinds table | +| `check_id` | Suite that produced the finding | +| `origin` | For the `model` and `scenarios` suites, which run once per entity, the entity whose pass produced the finding; absent for store-wide suites | +| `before` | A representative occurrence before the change, when one existed | +| `after` | A representative occurrence after the change, when one exists | +| `occurrences` | How many occurrences this change accounts for | +| `occurrences_before` | Occurrences of this identity before the change (for a successor, of the predecessor) | +| `occurrences_after` | Occurrences of this identity after the change | +| `reason` | For `KIND_UNKNOWN`, why the relevant scan cannot vouch for the finding | + +## Kinds + +| Kind | `before` | `after` | Meaning | +| --- | --- | --- | --- | +| `KIND_INTRODUCED` | absent | set | The change creates the finding; the before-scan's suite completed | +| `KIND_RESOLVED` | set | absent | The change clears the finding; the after-scan's suite completed | +| `KIND_SEVERITY_CHANGED` | set | set | Same finding, different severity | +| `KIND_SUBJECT_REMOVED` | set | absent | The finding's subject or origin entity no longer exists after the change; reported even when the after-scan's suite did not complete | +| `KIND_CARRIED_TO_SUCCESSOR` | set | set | The finding moved to a newer version of its subject or origin, with no version going down | +| `KIND_UNKNOWN` | either | either | A finding appeared or disappeared, but the scan on the other side did not complete its suite | + +For `KIND_UNKNOWN`, `before` alone means a resolution is unproven, and +`after` alone means novelty is unproven. + +## Finding identity + +Two occurrences are the same finding when they agree on suite, `origin` +(kind, namespace, slug and version), rule `code`, subject kind, namespace, +slug and version, `subject_field`, and +every other structured field of `ValidationIssue`. Message prose, severity, +`rule_title`, `rule_category` and `rule_default_severity` are excluded. + +## Coverage check ids + +The coverage of each scan lists these checks: + +| Check | Compared | Notes | +| --- | --- | --- | +| `store_hygiene` | yes | `PARTIAL` with `snapshot_truncated` when the snapshot hit the cap | +| `architecture` | yes | As above | +| `model` | yes | As above | +| `strategic` | yes | As above | +| `scenarios` | yes | As above | +| `specification_types` | no | `SKIPPED` with `delta_not_supported` when the world holds specifications, otherwise `kind_not_applicable` | +| `data_product_events` | no | `SKIPPED` with `delta_not_supported` when the world holds data products, otherwise `kind_not_applicable` | +| `data_flow` | no | Always `SKIPPED` with `analysis_not_compared` | +| `information_completeness` | no | Always `SKIPPED` with `analysis_not_compared` | + +A compared suite missing from a scan's coverage is treated as not +completed, with reason `suite_not_reported`. + +## Metrics + +| Metric | Instrument | Unit | +| --- | --- | --- | +| `trogon_atlas.consequence_delta.duration` | histogram | s | diff --git a/docs/reference/mcp-tools.md b/docs/reference/mcp-tools.md index 3cf1dd32..65c862b7 100644 --- a/docs/reference/mcp-tools.md +++ b/docs/reference/mcp-tools.md @@ -23,7 +23,7 @@ registered tool and its policy row disagree. | Tool | Minimum role | Effect | Idempotent | Profiles | gRPC methods | Parameters | Description | | --- | --- | --- | --- | --- | --- | --- | --- | | `accept_source_snapshot` | writer | destructive | no | `source-integration` | `AcceptSourceSnapshot` | `request_json` | AcceptSourceSnapshot: connected backend source control. Explicit selectors and ownership only. Returned scope, fidelity, incomplete and recovery state remain authoritative; preview is not an apply receipt. | -| `apply_manifests` | writer | write | yes | `authoring`, `repair` | `BatchGetEntities`, `BatchMutate` | `dry_run?`, `manifests_yaml`, `operation_id?` | Apply one or more YAML manifests (same format as trogon-atlas apply): create missing entities, update changed ones, skip unchanged ones, as one atomic batch. Set dry_run to validate without persisting. Non-blocking validation findings, strategic ones included, come back in the `issues` array as {severity, code, subject, field, message, op_index, op_entity} (op_index is the position among the changed manifests sent, op_entity the manifest's entity id): fix every warning and error. Each write is bound to the revision it was planned against: if another writer changed or created an entity in between, nothing is written and the call fails with data.category "stale_state" and data.next_action "replan", naming the entity with expected_etag and actual_etag (null means absent); call again to plan from fresh state. | +| `apply_manifests` | writer | write | yes | `authoring`, `repair` | `BatchGetEntities`, `BatchMutate` | `dry_run?`, `manifests_yaml`, `operation_id?` | Apply one or more YAML manifests (same format as trogon-atlas apply): create missing entities, update changed ones, skip unchanged ones, as one atomic batch. Set dry_run to validate without persisting. Non-blocking validation findings, strategic ones included, come back in the `issues` array as {severity, code, subject, field, message, op_index, op_entity} (op_index is the position among the changed manifests sent, op_entity the manifest's entity id): fix every warning and error. Each write is bound to the revision it was planned against: if another writer changed or created an entity in between, nothing is written and the call fails with data.category "stale_state" and data.next_action "replan", naming the entity with expected_etag and actual_etag (null means absent); call again to plan from fresh state. A dry run that sends a batch and has it validated also returns `consequence_delta`: which findings the change would introduce, resolve or leave unknown (advisory, never a refusal). It is absent when every manifest is unchanged (no batch is sent) or when the server could not compute it; an absent delta never means no findings changed. | | `apply_source_run` | writer | destructive | no | `source-integration` | `ApplySourceRun` | `request_json` | ApplySourceRun: connected backend source control. Explicit selectors and ownership only. Returned scope, fidelity, incomplete and recovery state remain authoritative; preview is not an apply receipt. | | `batch_get_entities` | reader | read | yes | `authoring`, `repair`, `review`, `source-integration`, `read-only` | `BatchGetEntities` | `refs` | Batch fetch entities by (kind, namespace, slug, version) tuples. Returns the entities in arbitrary order (missing entries omitted). | | `batch_mutate` | writer | destructive | yes | `repair` | `BatchMutate` | `operation_id?`, `ops`, `validate_only?` | Atomic batch of put/delete operations. Each op specifies op='put' (with entity_b64) or op='delete' (with kind/namespace/slug/version). Failure rolls back the whole batch and is reported as an MCP tool error whose data carries category/next_action (and code/next_action_detail when applicable), never as a success envelope with status=STATUS_FAILED inside it. Non-blocking validation findings, including strategic ones (BoundedContext, Domain, Subdomain rules), come back in the `issues` array next to binpb_base64, each tagged with op_index and op_entity of the put or delete op that produced it (validate_only dry runs included): fix every warning and error; an empty array means none. | @@ -103,7 +103,7 @@ registered tool and its policy row disagree. | `list_term_domain_declarations` | reader | read | yes | `authoring`, `repair`, `review`, `source-integration`, `read-only` | `ListTermDomainDeclarations` | `request_json` | ListTermDomainDeclarations: inspect native catalog declarations with exact selectors and backend checked scope. Incomplete or unavailable results do not prove absence; no approval or implementation evidence is inferred. | | `list_validation_rules` | reader | read | yes | `authoring`, `repair`, `review`, `source-integration`, `read-only` | `ListValidationRules` | `json?` | Static catalog of every validation rule the validator can emit (code, default severity, title, doc, subject_kind/subject_field hints, category). Use this to render legends, link findings to docs, or to power CI severity gates without reverse-engineering the validator. | | `list_versions` | reader | read | yes | `authoring`, `repair`, `review`, `source-integration`, `read-only` | `ListVersions` | `kind`, `namespace`, `slug` | List every stored version of (kind, namespace, slug), ordered ascending. | -| `merge_branch` | writer | destructive | no | `repair` | `MergeBranch` | `auto_merge?`, `dry_run?`, `json?`, `keep_branch?`, `name`, `operation_id?` | Merge a branch onto baseline (Phase 2: Review and merge). All-or-nothing: any conflict (see conflicts) or post-merge validation Error (see validation) persists nothing. Set dry_run to run full detection and validation without persisting even on success; the result splits the landable entries into added_count, changed_count and deleted_count and lists the post-merge findings, strategic ones included, in validation and, as JSON, in `issues` (Warning and Info findings are non-blocking; fix them). keep_branch controls whether a successful, non-dry-run merge deletes the branch afterward (default: deletes it). Set auto_merge to land the combined result for edit/edit conflicts whose two sides changed different fields. | +| `merge_branch` | writer | destructive | no | `repair` | `MergeBranch` | `auto_merge?`, `dry_run?`, `json?`, `keep_branch?`, `name`, `operation_id?` | Merge a branch onto baseline (Phase 2: Review and merge). All-or-nothing: any conflict (see conflicts) or post-merge validation Error (see validation) persists nothing. Set dry_run to run full detection and validation without persisting even on success; the result splits the landable entries into added_count, changed_count and deleted_count and lists the post-merge findings, strategic ones included, in validation and, as JSON, in `issues` (Warning and Info findings are non-blocking; fix them). keep_branch controls whether a successful, non-dry-run merge deletes the branch afterward (default: deletes it). Set auto_merge to land the combined result for edit/edit conflicts whose two sides changed different fields. A dry run that reaches validation also returns `consequence_delta`: which findings the change would introduce, resolve or leave unknown (advisory, never a refusal). It is absent when the merge has conflicts or when the server could not compute it; an absent delta never means no findings changed. | | `move_namespace` | admin | destructive | no | none | `MoveNamespace` | `id`, `parent` | Re-parent a namespace. Admin only: it is the one operation that crosses a tenancy boundary. Writes the registry row and nothing else, so no entity is re-keyed and no reference changes. | | `preview_source_run` | writer | destructive | no | `source-integration` | `PreviewSourceRun` | `request_json` | PreviewSourceRun: connected backend source control. Explicit selectors and ownership only. Returned scope, fidelity, incomplete and recovery state remain authoritative; preview is not an apply receipt. | | `protect_entity` | admin | write | yes | none | `ProtectEntity` | `kind`, `namespace`, `reason`, `slug`, `validate_only?` | Freeze every version of an identity (kind, namespace, slug), including versions not written yet, so every write route refuses it with ENTITY_PROTECTED until an admin lifts it. Admin only, baseline only, and the identity must already have a stored version. changed=false means it was already protected; protection then holds the existing row, whose reason may differ. | diff --git a/otel/registry/trogon_atlas/server/consequence_delta.yaml b/otel/registry/trogon_atlas/server/consequence_delta.yaml new file mode 100644 index 00000000..2920063f --- /dev/null +++ b/otel/registry/trogon_atlas/server/consequence_delta.yaml @@ -0,0 +1,15 @@ +file_format: definition/2 +metrics: + - name: trogon_atlas.consequence_delta.duration + requirement_level: recommended + instrument: histogram + unit: s + stability: stable + brief: Time spent scanning both worlds and comparing their findings for a consequence delta. + annotations: + code_generation: + metric_value_type: double + aggregation: + method: explicithistogram + parameters: + boundaries: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10] diff --git a/proto/trogonatlas/api/eventmodel/v1alpha1/service.proto b/proto/trogonatlas/api/eventmodel/v1alpha1/service.proto index a96b7f7a..b46eb68e 100644 --- a/proto/trogonatlas/api/eventmodel/v1alpha1/service.proto +++ b/proto/trogonatlas/api/eventmodel/v1alpha1/service.proto @@ -271,6 +271,9 @@ message PutEntityResponse { // `validation` means only the suites listed here ran, not a full model // check. ValidationCoverage coverage = 6; + // How the change would move the model's findings. Set only on + // `validate_only`; advisory, never a refusal. + ConsequenceDelta consequence_delta = 7; } message DeleteEntityRequest { @@ -417,6 +420,9 @@ message BatchMutateResponse { // referrer gate). Never invent coverage client-side; empty per-op // `validation` is not a full-model clean bill of health. ValidationCoverage coverage = 7; + // How the batch would move the model's findings. Set only on + // STATUS_VALIDATED; advisory, never a refusal. + ConsequenceDelta consequence_delta = 8; } // Bulk delete by structural query, replacing the "list every entity then @@ -1163,6 +1169,9 @@ message MergeBranchResponse { // not run full `model` or write-path `scenarios`; coverage reports that // honestly so empty `validation` is not read as a clean full-model gate. ValidationCoverage coverage = 9; + // How landing the branch would move baseline's findings. Set only on + // `dry_run` once validation was reached; advisory, never a refusal. + ConsequenceDelta consequence_delta = 10; } message UpdateBranchRequest { diff --git a/proto/trogonatlas/api/eventmodel/v1alpha1/validation.proto b/proto/trogonatlas/api/eventmodel/v1alpha1/validation.proto index 71452458..72dde1a2 100644 --- a/proto/trogonatlas/api/eventmodel/v1alpha1/validation.proto +++ b/proto/trogonatlas/api/eventmodel/v1alpha1/validation.proto @@ -226,3 +226,75 @@ message ValidationCoverage { // May be narrower than `effective_scope` for store-wide checks. ValidationScope requested_scope = 4; } + +// ---------- Consequence delta ------------------------------------------------ +// +// What a proposed change would do to the findings of the model it lands in. +// Both worlds are scanned with the same suites over the caller's whole +// visible snapshot, then compared by finding identity. Advisory only: a +// delta never refuses a write. See docs/explanation/consequence-delta.md. + +message FindingChange { + enum Kind { + KIND_UNSPECIFIED = 0; + // Present after the change and absent before, with the before scan + // complete for its suite. + KIND_INTRODUCED = 1; + // Present before and absent after, with the after scan complete for its + // suite and the subject still present. + KIND_RESOLVED = 2; + // Same identity on both sides with a different severity. + KIND_SEVERITY_CHANGED = 3; + // The change removes the finding's subject or origin, so the finding + // goes with it. + KIND_SUBJECT_REMOVED = 4; + // The finding now also appears on a newer version of its subject: the + // successor inherited it rather than introducing a new problem. + KIND_CARRIED_TO_SUCCESSOR = 5; + // Seen on one side only, but the other side's scan could not cover the + // finding's suite. `before` set alone: resolution is unproven. `after` + // set alone: novelty is unproven. `reason` names the gap. + KIND_UNKNOWN = 6; + } + Kind kind = 1; + // Suite the finding belongs to, as on ValidationCheckResult.check_id. + string check_id = 2; + // Representative issue on each side. Unset on the side where the + // identity has no occurrence. + ValidationIssue before = 3; + ValidationIssue after = 4; + // Occurrences of this identity this change accounts for. + uint32 occurrences = 5; + // Total occurrences of the identity in each world, so repeated findings + // are visible as counts rather than as duplicate entries. + uint32 occurrences_before = 6; + uint32 occurrences_after = 7; + // Why the change is KIND_UNKNOWN. Empty for every other kind. + string reason = 8; + // The entity whose validation pass produced the finding, for suites that + // run once per entity (`model`, `scenarios`). Part of the finding's + // identity: one rule on one subject reported by two models is two + // findings. Absent for suites that scan the whole store. + trogonatlas.eventmodel.v1alpha1.EntityRef origin = 9; +} + +message ConsequenceDelta { + // Every identity whose occurrences or severity differ, sorted by suite, + // code and subject. + repeated FindingChange changes = 1; + // Occurrences present on both sides with the same severity. + uint32 unchanged_count = 2; + // Coverage of the scan over the world before the change. + ValidationCoverage before_coverage = 3; + // Coverage of the scan over the candidate world. + ValidationCoverage after_coverage = 4; + // True only when every suite the delta compares completed on both sides + // and no applicable suite was left out. False never means "no change"; + // it means some finding may be missing from `changes`. + bool complete = 5; + // Snapshot identity of the visible world the change was compared + // against, computed like GetSnapshotId with no namespace filter. + string before_snapshot_id = 6; + // Snapshot identity of the candidate world the change would produce. + string after_snapshot_id = 7; +} diff --git a/rsworkspace/crates/trogon-atlas-cli/src/main.rs b/rsworkspace/crates/trogon-atlas-cli/src/main.rs index 72d946e3..8f71cdce 100644 --- a/rsworkspace/crates/trogon-atlas-cli/src/main.rs +++ b/rsworkspace/crates/trogon-atlas-cli/src/main.rs @@ -1203,6 +1203,7 @@ async fn run( "issues": outcome.issues.iter().map(apply::issue_to_json).collect::>(), }); apply::insert_coverage(&mut data, outcome.coverage.as_ref()); + apply::insert_consequence_delta(&mut data, outcome.consequence_delta.as_ref()); Ok((ExitCode::SUCCESS, Some(data))) } Cmd::Diff { filenames } => { @@ -2658,6 +2659,7 @@ async fn run( "validation": resp.validation.iter().map(apply::issue_to_json).collect::>(), }); apply::insert_coverage(&mut data, resp.coverage.as_ref()); + apply::insert_consequence_delta(&mut data, resp.consequence_delta.as_ref()); Ok((ExitCode::SUCCESS, Some(data))) } Status::Validated => { @@ -2687,6 +2689,7 @@ async fn run( "validation": resp.validation.iter().map(apply::issue_to_json).collect::>(), }); apply::insert_coverage(&mut data, resp.coverage.as_ref()); + apply::insert_consequence_delta(&mut data, resp.consequence_delta.as_ref()); Ok((ExitCode::SUCCESS, Some(data))) } Status::Conflicts => { @@ -2722,6 +2725,7 @@ async fn run( "validation": resp.validation.iter().map(apply::issue_to_json).collect::>(), }); apply::insert_coverage(&mut data, resp.coverage.as_ref()); + apply::insert_consequence_delta(&mut data, resp.consequence_delta.as_ref()); Ok((ExitCode::from(EXIT_FINDINGS), Some(data))) } Status::Unspecified => { diff --git a/rsworkspace/crates/trogon-atlas-client/src/apply.rs b/rsworkspace/crates/trogon-atlas-client/src/apply.rs index e82bf53b..5208b974 100644 --- a/rsworkspace/crates/trogon-atlas-client/src/apply.rs +++ b/rsworkspace/crates/trogon-atlas-client/src/apply.rs @@ -344,6 +344,51 @@ pub struct ApplyOutcome { /// or the server omitted it (for example operation_id replay). Never /// invent coverage client-side. pub coverage: Option, + /// How a dry run would move the model's findings. Present only for a + /// dry run that reached validation. + pub consequence_delta: Option, +} + +/// Adds a server-reported consequence delta to a JSON object. An absent +/// delta stays absent: no delta is not the same as no change. +pub fn insert_consequence_delta(body: &mut Value, delta: Option<&pb::ConsequenceDelta>) { + if let (Some(fields), Some(delta)) = (body.as_object_mut(), delta) { + fields.insert("consequence_delta".into(), consequence_delta_to_json(delta)); + } +} + +/// JSON shape for [`pb::ConsequenceDelta`], for CLI/MCP structured output. +#[must_use] +pub fn consequence_delta_to_json(delta: &pb::ConsequenceDelta) -> Value { + let finding = |issue: Option<&pb::ValidationIssue>| { + issue.map(|issue| { + let mut body = issue_to_json(issue); + if let Some(fields) = body.as_object_mut() { + fields.insert("subject_field".into(), issue.subject_field.clone().into()); + } + body + }) + }; + json!({ + "complete": delta.complete, + "unchanged_count": delta.unchanged_count, + "before_snapshot_id": delta.before_snapshot_id, + "after_snapshot_id": delta.after_snapshot_id, + "before_coverage": delta.before_coverage.as_ref().map(coverage_to_json), + "after_coverage": delta.after_coverage.as_ref().map(coverage_to_json), + "changes": delta.changes.iter().map(|c| json!({ + "kind": pb::finding_change::Kind::try_from(c.kind) + .map_or("KIND_UNSPECIFIED", |k| k.as_str_name()), + "check_id": c.check_id, + "origin": ref_string(c.origin.as_ref()), + "before": finding(c.before.as_ref()), + "after": finding(c.after.as_ref()), + "occurrences": c.occurrences, + "occurrences_before": c.occurrences_before, + "occurrences_after": c.occurrences_after, + "reason": c.reason, + })).collect::>(), + }) } /// Adds server-reported coverage to a JSON object. Absent coverage stays @@ -436,6 +481,7 @@ pub async fn apply( let mut op_issues = Vec::new(); let mut operation_receipt = None; let mut coverage = None; + let mut consequence_delta = None; if !changed.is_empty() { let ops = changed .iter() @@ -507,6 +553,7 @@ pub async fn apply( .flat_map(|o| o.issues.iter().cloned()) .collect(); coverage = resp.coverage; + consequence_delta = resp.consequence_delta; } let lines = plan @@ -519,6 +566,7 @@ pub async fn apply( op_issues, operation_receipt, coverage, + consequence_delta, }) } diff --git a/rsworkspace/crates/trogon-atlas-mcp/src/server.rs b/rsworkspace/crates/trogon-atlas-mcp/src/server.rs index 864697e9..58e3619f 100644 --- a/rsworkspace/crates/trogon-atlas-mcp/src/server.rs +++ b/rsworkspace/crates/trogon-atlas-mcp/src/server.rs @@ -1588,6 +1588,24 @@ fn with_coverage( result } +/// Add a server-reported consequence delta to an already-built JSON tool +/// result body. An absent delta leaves the body untouched. +fn with_consequence_delta( + mut result: CallToolResult, + delta: Option<&pb::ConsequenceDelta>, +) -> CallToolResult { + let body = result + .content + .first() + .and_then(|block| block.as_text()) + .and_then(|text| serde_json::from_str::(&text.text).ok()); + if let (Some(mut body), Some(_)) = (body, delta) { + trogon_atlas_client::apply::insert_consequence_delta(&mut body, delta); + result.content[0] = ContentBlock::text(body.to_string()); + } + result +} + fn with_issues(mut result: CallToolResult, issues: Vec) -> CallToolResult { let body = result .content @@ -4152,10 +4170,13 @@ impl McpServer { .map_err(tool_err("put_entity")) .map_err(|e| attach_operation_id(e, &operation_id))? .into_inner(); - Ok(self.scoped_envelope_with_issues( - &resp, - &issues_json(&resp.validation), - resp.coverage.as_ref(), + Ok(with_consequence_delta( + self.scoped_envelope_with_issues( + &resp, + &issues_json(&resp.validation), + resp.coverage.as_ref(), + ), + resp.consequence_delta.as_ref(), )) } @@ -4243,10 +4264,13 @@ impl McpServer { .map_err(tool_err("put_entity_json")) .map_err(|e| attach_operation_id(e, &operation_id))? .into_inner(); - Ok(self.scoped_envelope_with_issues( - &resp, - &issues_json(&resp.validation), - resp.coverage.as_ref(), + Ok(with_consequence_delta( + self.scoped_envelope_with_issues( + &resp, + &issues_json(&resp.validation), + resp.coverage.as_ref(), + ), + resp.consequence_delta.as_ref(), )) } @@ -4293,10 +4317,13 @@ impl McpServer { if resp.status == pb::batch_mutate_response::Status::Failed as i32 { return Err(McpError::from(batch_mutate_failure(&resp)).into()); } - Ok(self.scoped_envelope_with_issues( - &resp, - &batch_issues_json(&op_entities, &resp), - resp.coverage.as_ref(), + Ok(with_consequence_delta( + self.scoped_envelope_with_issues( + &resp, + &batch_issues_json(&op_entities, &resp), + resp.coverage.as_ref(), + ), + resp.consequence_delta.as_ref(), )) } @@ -4468,10 +4495,13 @@ impl McpServer { let err: rmcp::ErrorData = McpError::from(batch_mutate_failure(&resp)).into(); return Err(attach_operation_id(err, &operation_id)); } - Ok(self.scoped_envelope_with_issues( - &resp, - &batch_issues_json(&op_entities, &resp), - resp.coverage.as_ref(), + Ok(with_consequence_delta( + self.scoped_envelope_with_issues( + &resp, + &batch_issues_json(&op_entities, &resp), + resp.coverage.as_ref(), + ), + resp.consequence_delta.as_ref(), )) } @@ -4997,7 +5027,7 @@ impl McpServer { } #[tool( - description = "Apply one or more YAML manifests (same format as trogon-atlas apply): create missing entities, update changed ones, skip unchanged ones, as one atomic batch. Set dry_run to validate without persisting. Non-blocking validation findings, strategic ones included, come back in the `issues` array as {severity, code, subject, field, message, op_index, op_entity} (op_index is the position among the changed manifests sent, op_entity the manifest's entity id): fix every warning and error. Each write is bound to the revision it was planned against: if another writer changed or created an entity in between, nothing is written and the call fails with data.category \"stale_state\" and data.next_action \"replan\", naming the entity with expected_etag and actual_etag (null means absent); call again to plan from fresh state.", + description = "Apply one or more YAML manifests (same format as trogon-atlas apply): create missing entities, update changed ones, skip unchanged ones, as one atomic batch. Set dry_run to validate without persisting. Non-blocking validation findings, strategic ones included, come back in the `issues` array as {severity, code, subject, field, message, op_index, op_entity} (op_index is the position among the changed manifests sent, op_entity the manifest's entity id): fix every warning and error. Each write is bound to the revision it was planned against: if another writer changed or created an entity in between, nothing is written and the call fails with data.category \"stale_state\" and data.next_action \"replan\", naming the entity with expected_etag and actual_etag (null means absent); call again to plan from fresh state. A dry run that sends a batch and has it validated also returns `consequence_delta`: which findings the change would introduce, resolve or leave unknown (advisory, never a refusal). It is absent when every manifest is unchanged (no batch is sent) or when the server could not compute it; an absent delta never means no findings changed.", annotations( read_only_hint = false, destructive_hint = false, @@ -5041,6 +5071,10 @@ impl McpServer { "operation_id": (!operation_id.is_empty()).then_some(operation_id.as_str()), }); trogon_atlas_client::apply::insert_coverage(&mut body, outcome.coverage.as_ref()); + trogon_atlas_client::apply::insert_consequence_delta( + &mut body, + outcome.consequence_delta.as_ref(), + ); Ok(CallToolResult::success(vec![ContentBlock::text( body.to_string(), )])) @@ -5232,7 +5266,7 @@ impl McpServer { } #[tool( - description = "Merge a branch onto baseline (Phase 2: Review and merge). All-or-nothing: any conflict (see conflicts) or post-merge validation Error (see validation) persists nothing. Set dry_run to run full detection and validation without persisting even on success; the result splits the landable entries into added_count, changed_count and deleted_count and lists the post-merge findings, strategic ones included, in validation and, as JSON, in `issues` (Warning and Info findings are non-blocking; fix them). keep_branch controls whether a successful, non-dry-run merge deletes the branch afterward (default: deletes it). Set auto_merge to land the combined result for edit/edit conflicts whose two sides changed different fields.", + description = "Merge a branch onto baseline (Phase 2: Review and merge). All-or-nothing: any conflict (see conflicts) or post-merge validation Error (see validation) persists nothing. Set dry_run to run full detection and validation without persisting even on success; the result splits the landable entries into added_count, changed_count and deleted_count and lists the post-merge findings, strategic ones included, in validation and, as JSON, in `issues` (Warning and Info findings are non-blocking; fix them). keep_branch controls whether a successful, non-dry-run merge deletes the branch afterward (default: deletes it). Set auto_merge to land the combined result for edit/edit conflicts whose two sides changed different fields. A dry run that reaches validation also returns `consequence_delta`: which findings the change would introduce, resolve or leave unknown (advisory, never a refusal). It is absent when the merge has conflicts or when the server could not compute it; an absent delta never means no findings changed.", annotations( read_only_hint = false, destructive_hint = true, @@ -5259,16 +5293,19 @@ impl McpServer { .map_err(tool_err("merge_branch")) .map_err(|e| attach_operation_id(e, &operation_id))? .into_inner(); - Ok(with_coverage( - with_issues( - ok_response( - "trogonatlas.api.eventmodel.v1alpha1.MergeBranchResponse", - &resp, - p.json.unwrap_or(false), + Ok(with_consequence_delta( + with_coverage( + with_issues( + ok_response( + "trogonatlas.api.eventmodel.v1alpha1.MergeBranchResponse", + &resp, + p.json.unwrap_or(false), + ), + issues_json(&resp.validation), ), - issues_json(&resp.validation), + resp.coverage.as_ref(), ), - resp.coverage.as_ref(), + resp.consequence_delta.as_ref(), )) } diff --git a/rsworkspace/crates/trogon-atlas-mcp/tests/mcp_integration.rs b/rsworkspace/crates/trogon-atlas-mcp/tests/mcp_integration.rs index 2db2ab98..b22b3b8c 100644 --- a/rsworkspace/crates/trogon-atlas-mcp/tests/mcp_integration.rs +++ b/rsworkspace/crates/trogon-atlas-mcp/tests/mcp_integration.rs @@ -2894,3 +2894,36 @@ async fn association_mcp_round_trips_attachments_and_parameter_maps_diffs_real_c ); service.cancel().await.unwrap(); } + +#[tokio::test(flavor = "multi_thread")] +async fn put_entity_json_dry_run_reports_the_consequence_delta() { + use serde_json::json; + let (client, service, _) = start_mcp_client().await; + let model = |members: Vec| { + json!({"eventModel":{"id":{"namespace":"mcp-delta","slug":"checkout","version":"1"},"title":"Checkout","members":members}}).to_string() + }; + call_ok( + &client, + "put_entity_json", + put_entity_json_args(&model(vec![])), + ) + .await; + let mut preview = put_entity_json_args(&model(vec![ + json!({"kind":"ENTITY_KIND_EVENT","id":{"namespace":"mcp-delta","slug":"missing","version":"1"}}), + ])); + preview.insert("validate_only".into(), json!(true)); + + let body = call_tool_result_text(&call_ok(&client, "put_entity_json", preview).await); + + let delta = &body["consequence_delta"]; + assert_eq!(delta["complete"], json!(true), "{body}"); + let introduced = delta["changes"] + .as_array() + .unwrap() + .iter() + .find(|change| change["after"]["code"] == "MEMBER_NOT_FOUND") + .unwrap_or_else(|| panic!("{body}")); + assert_eq!(introduced["kind"], "KIND_INTRODUCED"); + assert_eq!(introduced["origin"], "event_model:mcp-delta/checkout@1"); + service.cancel().await.unwrap(); +} diff --git a/rsworkspace/crates/trogon-atlas-proto/src/gen/trogonatlas.api.eventmodel.v1alpha1.rs b/rsworkspace/crates/trogon-atlas-proto/src/gen/trogonatlas.api.eventmodel.v1alpha1.rs index aa888f40..53c73653 100644 --- a/rsworkspace/crates/trogon-atlas-proto/src/gen/trogonatlas.api.eventmodel.v1alpha1.rs +++ b/rsworkspace/crates/trogon-atlas-proto/src/gen/trogonatlas.api.eventmodel.v1alpha1.rs @@ -2082,6 +2082,129 @@ pub struct ValidationCoverage { #[prost(message, optional, tag="4")] pub requested_scope: ::core::option::Option, } +// ---------- Consequence delta ------------------------------------------------ +// +// What a proposed change would do to the findings of the model it lands in. +// Both worlds are scanned with the same suites over the caller's whole +// visible snapshot, then compared by finding identity. Advisory only: a +// delta never refuses a write. See docs/explanation/consequence-delta.md. + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct FindingChange { + #[prost(enumeration="finding_change::Kind", tag="1")] + pub kind: i32, + /// Suite the finding belongs to, as on ValidationCheckResult.check_id. + #[prost(string, tag="2")] + pub check_id: ::prost::alloc::string::String, + /// Representative issue on each side. Unset on the side where the + /// identity has no occurrence. + #[prost(message, optional, tag="3")] + pub before: ::core::option::Option, + #[prost(message, optional, tag="4")] + pub after: ::core::option::Option, + /// Occurrences of this identity this change accounts for. + #[prost(uint32, tag="5")] + pub occurrences: u32, + /// Total occurrences of the identity in each world, so repeated findings + /// are visible as counts rather than as duplicate entries. + #[prost(uint32, tag="6")] + pub occurrences_before: u32, + #[prost(uint32, tag="7")] + pub occurrences_after: u32, + /// Why the change is KIND_UNKNOWN. Empty for every other kind. + #[prost(string, tag="8")] + pub reason: ::prost::alloc::string::String, + /// The entity whose validation pass produced the finding, for suites that + /// run once per entity (`model`, `scenarios`). Part of the finding's + /// identity: one rule on one subject reported by two models is two + /// findings. Absent for suites that scan the whole store. + #[prost(message, optional, tag="9")] + pub origin: ::core::option::Option, +} +/// Nested message and enum types in `FindingChange`. +pub mod finding_change { + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum Kind { + Unspecified = 0, + /// Present after the change and absent before, with the before scan + /// complete for its suite. + Introduced = 1, + /// Present before and absent after, with the after scan complete for its + /// suite and the subject still present. + Resolved = 2, + /// Same identity on both sides with a different severity. + SeverityChanged = 3, + /// The change removes the finding's subject or origin, so the finding + /// goes with it. + SubjectRemoved = 4, + /// The finding now also appears on a newer version of its subject: the + /// successor inherited it rather than introducing a new problem. + CarriedToSuccessor = 5, + /// Seen on one side only, but the other side's scan could not cover the + /// finding's suite. `before` set alone: resolution is unproven. `after` + /// set alone: novelty is unproven. `reason` names the gap. + Unknown = 6, + } + impl Kind { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Unspecified => "KIND_UNSPECIFIED", + Self::Introduced => "KIND_INTRODUCED", + Self::Resolved => "KIND_RESOLVED", + Self::SeverityChanged => "KIND_SEVERITY_CHANGED", + Self::SubjectRemoved => "KIND_SUBJECT_REMOVED", + Self::CarriedToSuccessor => "KIND_CARRIED_TO_SUCCESSOR", + Self::Unknown => "KIND_UNKNOWN", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "KIND_UNSPECIFIED" => Some(Self::Unspecified), + "KIND_INTRODUCED" => Some(Self::Introduced), + "KIND_RESOLVED" => Some(Self::Resolved), + "KIND_SEVERITY_CHANGED" => Some(Self::SeverityChanged), + "KIND_SUBJECT_REMOVED" => Some(Self::SubjectRemoved), + "KIND_CARRIED_TO_SUCCESSOR" => Some(Self::CarriedToSuccessor), + "KIND_UNKNOWN" => Some(Self::Unknown), + _ => None, + } + } + } +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ConsequenceDelta { + /// Every identity whose occurrences or severity differ, sorted by suite, + /// code and subject. + #[prost(message, repeated, tag="1")] + pub changes: ::prost::alloc::vec::Vec, + /// Occurrences present on both sides with the same severity. + #[prost(uint32, tag="2")] + pub unchanged_count: u32, + /// Coverage of the scan over the world before the change. + #[prost(message, optional, tag="3")] + pub before_coverage: ::core::option::Option, + /// Coverage of the scan over the candidate world. + #[prost(message, optional, tag="4")] + pub after_coverage: ::core::option::Option, + /// True only when every suite the delta compares completed on both sides + /// and no applicable suite was left out. False never means "no change"; + /// it means some finding may be missing from `changes`. + #[prost(bool, tag="5")] + pub complete: bool, + /// Snapshot identity of the visible world the change was compared + /// against, computed like GetSnapshotId with no namespace filter. + #[prost(string, tag="6")] + pub before_snapshot_id: ::prost::alloc::string::String, + /// Snapshot identity of the candidate world the change would produce. + #[prost(string, tag="7")] + pub after_snapshot_id: ::prost::alloc::string::String, +} // Tenant type libraries are stored as proto source text and compiled on // demand. These messages let a client check a library and look up the // compiled types before it writes anything that depends on them. Every RPC @@ -4042,6 +4165,10 @@ pub struct PutEntityResponse { /// check. #[prost(message, optional, tag="6")] pub coverage: ::core::option::Option, + /// How the change would move the model's findings. Set only on + /// `validate_only`; advisory, never a refusal. + #[prost(message, optional, tag="7")] + pub consequence_delta: ::core::option::Option, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteEntityRequest { @@ -4238,6 +4365,10 @@ pub struct BatchMutateResponse { /// `validation` is not a full-model clean bill of health. #[prost(message, optional, tag="7")] pub coverage: ::core::option::Option, + /// How the batch would move the model's findings. Set only on + /// STATUS_VALIDATED; advisory, never a refusal. + #[prost(message, optional, tag="8")] + pub consequence_delta: ::core::option::Option, } /// Nested message and enum types in `BatchMutateResponse`. pub mod batch_mutate_response { @@ -5216,6 +5347,10 @@ pub struct MergeBranchResponse { /// honestly so empty `validation` is not read as a clean full-model gate. #[prost(message, optional, tag="9")] pub coverage: ::core::option::Option, + /// How landing the branch would move baseline's findings. Set only on + /// `dry_run` once validation was reached; advisory, never a refusal. + #[prost(message, optional, tag="10")] + pub consequence_delta: ::core::option::Option, } /// Nested message and enum types in `MergeBranchResponse`. pub mod merge_branch_response { diff --git a/rsworkspace/crates/trogon-atlas-semconv/src/semconv/metric.rs b/rsworkspace/crates/trogon-atlas-semconv/src/semconv/metric.rs index aa09c70e..46665765 100644 --- a/rsworkspace/crates/trogon-atlas-semconv/src/semconv/metric.rs +++ b/rsworkspace/crates/trogon-atlas-semconv/src/semconv/metric.rs @@ -923,6 +923,42 @@ impl Default for ChangesetsLostObservable { Self::new(&super::meter::Meter::default(), |_| {}) } } +/// Time spent scanning both worlds and comparing their findings for a consequence delta. +#[derive(Clone)] +pub struct ConsequenceDeltaDuration { + instrument: ::opentelemetry::metrics::Histogram, +} + +impl ConsequenceDeltaDuration { + pub const NAME: &'static str = "trogon_atlas.consequence_delta.duration"; + pub const UNIT: &'static str = "s"; + pub const DESCRIPTION: &'static str = + "Time spent scanning both worlds and comparing their findings for a consequence delta."; + + pub fn new(meter: &super::meter::Meter) -> Self { + let instrument = meter + .inner() + .f64_histogram(Self::NAME) + .with_description(Self::DESCRIPTION) + .with_unit(Self::UNIT) + .with_boundaries(vec![ + 0.001_f64, 0.005_f64, 0.01_f64, 0.025_f64, 0.05_f64, 0.1_f64, 0.25_f64, 0.5_f64, + 1_f64, 2.5_f64, 5_f64, 10_f64, + ]) + .build(); + Self { instrument } + } + pub fn record(&self, value: f64) { + let attributes = []; + self.instrument.record(value, &attributes); + } +} + +impl Default for ConsequenceDeltaDuration { + fn default() -> Self { + Self::new(&super::meter::Meter::default()) + } +} /// Git mirror apply calls that failed. #[derive(Clone)] pub struct GitMirrorFailures { diff --git a/rsworkspace/crates/trogon-atlas-server/src/consequence.rs b/rsworkspace/crates/trogon-atlas-server/src/consequence.rs new file mode 100644 index 00000000..8a7fd94c --- /dev/null +++ b/rsworkspace/crates/trogon-atlas-server/src/consequence.rs @@ -0,0 +1,1103 @@ +//! Consequence deltas: how a candidate world moves the findings of the world +//! it would replace. +//! +//! Both worlds are scanned in full with the same suites, then compared by +//! `FindingIdentity`. A finding missing from one side is only called +//! introduced or resolved when that side's scan covered the finding's suite; +//! otherwise it is reported as unknown. See +//! `docs/explanation/consequence-delta.md`. + +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; + +use prost::Message as _; +use trogon_atlas_proto as pb; +use trogon_atlas_store::{ + refs::{entity_id, entity_kind}, + StoredEntity, +}; + +use crate::validation; + +/// Validation reached through stored type libraries and data-product event +/// lookups, which read the store rather than a candidate world. +pub const CHECK_SPECIFICATION_TYPES: &str = "specification_types"; +pub const CHECK_DATA_PRODUCT_EVENTS: &str = "data_product_events"; +/// Scoped analyses whose results may come from a language model, so two +/// runs over one world need not agree. +pub const CHECK_DATA_FLOW: &str = "data_flow"; +pub const CHECK_INFORMATION_COMPLETENESS: &str = "information_completeness"; + +/// An applicable suite the delta cannot evaluate against a candidate world. +pub const REASON_DELTA_NOT_SUPPORTED: &str = "delta_not_supported"; +/// A non-validation analysis the delta deliberately does not compare. +pub const REASON_ANALYSIS_NOT_COMPARED: &str = "analysis_not_compared"; +/// A compared suite the scan's coverage does not mention. +const REASON_SUITE_NOT_REPORTED: &str = "suite_not_reported"; + +/// Suites whose findings the delta compares. +const COMPARED_SUITES: [&str; 5] = [ + validation::CHECK_STORE_HYGIENE, + validation::CHECK_ARCHITECTURE, + validation::CHECK_MODEL, + validation::CHECK_STRATEGIC, + validation::CHECK_SCENARIOS, +]; + +/// The exact entity a finding is attached to, split so a successor version +/// of the same entity can be recognized. +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)] +struct Subject { + kind: i32, + namespace: String, + slug: String, + version: u64, +} + +impl Subject { + fn of(subject: &pb::EntityRef) -> Self { + let id = subject.id.clone().unwrap_or_default(); + Self { + kind: subject.kind, + namespace: id.namespace, + slug: id.slug, + version: id.version, + } + } + + fn of_stored(stored: &StoredEntity) -> Option { + let kind = entity_kind(&stored.entity)?; + Some(Self::of(&pb::EntityRef { + kind: kind as i32, + id: entity_id(&stored.entity).cloned(), + })) + } + + fn entity_ref(&self) -> pb::EntityRef { + pb::EntityRef { + kind: self.kind, + id: Some(pb::Id { + namespace: self.namespace.clone(), + slug: self.slug.clone(), + version: self.version, + }), + } + } + + fn key(&self) -> Option { + let kind = pb::EntityKind::try_from(self.kind).ok()?; + Some(pb::EntityKey { + kind, + namespace: self.namespace.clone(), + slug: self.slug.clone(), + version: self.version, + }) + } +} + +/// What makes two findings the same finding: suite, the entity whose pass +/// produced it, rule code, exact subject and version, field, and every +/// structured field of the issue that is not prose, severity or denormalized +/// rule context. Structured fields added to `ValidationIssue` later join the +/// identity without changes here. +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)] +struct FindingIdentity { + suite: &'static str, + origin: Option, + code: String, + subject: Option, + field: String, + structure: Vec, + /// `structure` with every entity version its evidence pins erased, + /// which a finding shares with the same finding on a successor. + unversioned: Vec, + /// The evidence versions `unversioned` erased, in encoding order. + evidence_versions: Vec, +} + +impl FindingIdentity { + fn of(finding: &Finding) -> Self { + let Finding { + suite, + origin, + issue, + } = finding; + let mut structure = issue.clone(); + structure.severity = 0; + structure.code.clear(); + structure.message.clear(); + structure.subject = None; + structure.subject_field.clear(); + structure.rule_title.clear(); + structure.rule_category.clear(); + structure.rule_default_severity = 0; + // Evidence detail is presentation only, like the message. + for evidence in &mut structure.evidence { + evidence.detail.clear(); + } + let mut unversioned = structure.clone(); + let mut evidence_versions = Vec::new(); + for evidence in &mut unversioned.evidence { + let pinned = [&mut evidence.entity, &mut evidence.target] + .into_iter() + .flatten() + .filter_map(|entity| entity.id.as_mut()) + .chain( + evidence + .occurrence + .as_mut() + .and_then(|occurrence| occurrence.storyboard.as_mut()), + ); + for id in pinned { + evidence_versions.push(id.version); + id.version = 0; + } + } + Self { + suite, + origin: origin.clone(), + code: issue.code.clone(), + subject: issue.subject.as_ref().map(Subject::of), + field: issue.subject_field.clone(), + structure: structure.encode_to_vec(), + unversioned: unversioned.encode_to_vec(), + evidence_versions, + } + } + + /// The identity with every entity version erased, and those versions. + fn lineage(&self) -> (Self, Versions) { + let version = |s: &Option| s.as_ref().map_or(0, |s| s.version); + let versions = Versions { + origin: version(&self.origin), + subject: version(&self.subject), + evidence: self.evidence_versions.clone(), + }; + let mut lineage = self.clone(); + lineage.structure.clone_from(&self.unversioned); + lineage.evidence_versions.clear(); + for s in [&mut lineage.origin, &mut lineage.subject] + .into_iter() + .flatten() + { + s.version = 0; + } + (lineage, versions) + } + + fn entities(&self) -> impl Iterator + '_ { + [&self.origin, &self.subject] + .into_iter() + .flatten() + .filter_map(Subject::key) + } +} + +/// Entity versions a finding is pinned to, its evidence's included. One +/// identity succeeds another when no version went down and at least one +/// went up. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +struct Versions { + origin: u64, + subject: u64, + evidence: Vec, +} + +impl Versions { + fn precedes(&self, other: &Self) -> bool { + self != other + && self.origin <= other.origin + && self.subject <= other.subject + && self.evidence.len() == other.evidence.len() + && self + .evidence + .iter() + .zip(&other.evidence) + .all(|(before, after)| before <= after) + } +} + +/// One occurrence of an issue, with the suite that reported it and, for +/// suites that run once per entity, the entity the pass ran over. +#[derive(Debug, Clone)] +struct Finding { + suite: &'static str, + origin: Option, + issue: pb::ValidationIssue, +} + +/// One world's findings, tagged by suite, with the coverage the scan earned +/// and the entity keys it held. +#[derive(Debug, Clone)] +pub struct Scan { + findings: Vec, + coverage: pb::ValidationCoverage, + keys: HashSet, +} + +impl Scan { + /// Every compared suite over the whole of `world`. Architecture resources + /// resolve against `world` plus `architecture_context`. + #[must_use] + pub fn of( + world: &[StoredEntity], + architecture_context: &[StoredEntity], + snapshot_truncated: bool, + scope: &pb::ValidationScope, + ) -> Self { + let mut findings: Vec = Vec::new(); + let mut tag = |suite: &'static str, + origin: Option<&StoredEntity>, + issues: Vec| { + let origin = origin.and_then(Subject::of_stored); + findings.extend(issues.into_iter().map(|issue| Finding { + suite, + origin: origin.clone(), + issue, + })); + }; + tag( + validation::CHECK_STORE_HYGIENE, + None, + validation::validate_store_hygiene(world), + ); + tag( + validation::CHECK_ARCHITECTURE, + None, + validation::validate_architecture_resources(world, architecture_context), + ); + let mut models = 0usize; + let mut slices = 0usize; + let mut namespaces: BTreeSet<&str> = BTreeSet::new(); + let mut has_specifications = false; + let mut has_data_products = false; + for stored in world { + if let Some(id) = entity_id(&stored.entity) { + namespaces.insert(id.namespace.as_str()); + } + match stored.entity.kind.as_ref() { + Some(pb::entity::Kind::EventModel(model)) => { + models += 1; + tag( + validation::CHECK_MODEL, + Some(stored), + validation::validate_model_content(model, world), + ); + } + Some( + pb::entity::Kind::CommandSlice(_) + | pb::entity::Kind::ReadModelSlice(_) + | pb::entity::Kind::AutomationSlice(_), + ) => { + slices += 1; + tag( + validation::CHECK_SCENARIOS, + Some(stored), + validation::validate_scenarios(&stored.entity, world), + ); + } + Some(pb::entity::Kind::Specification(_)) => has_specifications = true, + Some(pb::entity::Kind::DataProduct(_)) => has_data_products = true, + _ => {} + } + } + tag( + validation::CHECK_STRATEGIC, + None, + validation::strategic_chain_issues(namespaces.iter().copied(), world), + ); + + let skipped = |check_id: &str, reason: &str| { + validation::coverage_check( + check_id, + pb::validation_check_result::State::Skipped, + reason, + 0, + ) + }; + let store_backed = |check_id: &str, applies: bool| { + skipped( + check_id, + if applies { + REASON_DELTA_NOT_SUPPORTED + } else { + validation::REASON_KIND_NOT_APPLICABLE + }, + ) + }; + let count = |n: usize| i32::try_from(n).unwrap_or(i32::MAX); + let checks = vec![ + validation::ran_against_snapshot( + validation::CHECK_STORE_HYGIENE, + 1, + snapshot_truncated, + ), + validation::ran_against_snapshot(validation::CHECK_ARCHITECTURE, 1, snapshot_truncated), + validation::ran_against_snapshot( + validation::CHECK_MODEL, + count(models), + snapshot_truncated, + ), + validation::ran_against_snapshot( + validation::CHECK_STRATEGIC, + count(namespaces.len()), + snapshot_truncated, + ), + validation::ran_against_snapshot( + validation::CHECK_SCENARIOS, + count(slices), + snapshot_truncated, + ), + store_backed(CHECK_SPECIFICATION_TYPES, has_specifications), + store_backed(CHECK_DATA_PRODUCT_EVENTS, has_data_products), + skipped(CHECK_DATA_FLOW, REASON_ANALYSIS_NOT_COMPARED), + skipped(CHECK_INFORMATION_COMPLETENESS, REASON_ANALYSIS_NOT_COMPARED), + ]; + let keys = world.iter().filter_map(stored_key).collect(); + Self { + findings, + coverage: validation::validation_coverage( + checks, + scope.clone(), + scope.clone(), + snapshot_truncated, + ), + keys, + } + } + + /// Why this scan cannot vouch for `suite`, or `None` when it completed. + fn gap(&self, suite: &str) -> Option { + let Some(check) = self.coverage.checks.iter().find(|c| c.check_id == suite) else { + return Some(REASON_SUITE_NOT_REPORTED.to_owned()); + }; + if check.state == pb::validation_check_result::State::Completed as i32 { + return None; + } + Some(if check.reason.is_empty() { + pb::validation_check_result::State::try_from(check.state) + .map_or("STATE_UNSPECIFIED", |s| s.as_str_name()) + .to_owned() + } else { + check.reason.clone() + }) + } + + fn leaves_out_applicable_suites(&self) -> bool { + self.coverage + .checks + .iter() + .any(|c| c.reason == REASON_DELTA_NOT_SUPPORTED) + } + + fn grouped(&self) -> BTreeMap> { + let mut groups: BTreeMap> = BTreeMap::new(); + for finding in &self.findings { + groups + .entry(FindingIdentity::of(finding)) + .or_default() + .push(&finding.issue); + } + groups + } +} + +fn stored_key(stored: &StoredEntity) -> Option { + Some(pb::EntityKey::new( + entity_kind(&stored.entity)?, + entity_id(&stored.entity)?, + )) +} + +/// Entities of `context` whose keys neither world holds. A context entry for +/// a key the worlds do hold is either a duplicate or, after the change, a +/// stale copy of something the change replaced or removed. +#[must_use] +pub fn context_outside( + context: Vec, + worlds: &[&[StoredEntity]], +) -> Vec { + let held: HashSet = worlds + .iter() + .flat_map(|world| world.iter().filter_map(stored_key)) + .collect(); + context + .into_iter() + .filter(|stored| stored_key(stored).is_none_or(|key| !held.contains(&key))) + .collect() +} + +fn occurrences(n: usize) -> u32 { + u32::try_from(n).unwrap_or(u32::MAX) +} + +struct Pending<'a> { + identity: FindingIdentity, + issues: Vec<&'a pb::ValidationIssue>, +} + +/// Compare two scans of the same scope. The snapshot ids bind the report to +/// the worlds it was computed from. +#[must_use] +pub fn compare( + before: &Scan, + after: &Scan, + before_snapshot_id: String, + after_snapshot_id: String, +) -> pb::ConsequenceDelta { + use pb::finding_change::Kind; + + let before_groups = before.grouped(); + let after_groups = after.grouped(); + let identities: BTreeSet<&FindingIdentity> = + before_groups.keys().chain(after_groups.keys()).collect(); + let total = |groups: &BTreeMap>, + identity: &FindingIdentity| { + occurrences(groups.get(identity).map_or(0, Vec::len)) + }; + + let mut changes: Vec<(FindingIdentity, pb::FindingChange)> = Vec::new(); + let mut unchanged = 0usize; + let mut gone: Vec = Vec::new(); + let mut new: Vec = Vec::new(); + for identity in identities { + let mut was: Vec<&pb::ValidationIssue> = + before_groups.get(identity).cloned().unwrap_or_default(); + let mut now: Vec<&pb::ValidationIssue> = + after_groups.get(identity).cloned().unwrap_or_default(); + was.retain(|issue| { + match now + .iter() + .position(|other| other.severity == issue.severity) + { + Some(at) => { + now.remove(at); + unchanged += 1; + false + } + None => true, + } + }); + let shifted = was.len().min(now.len()); + if shifted > 0 { + changes.push(( + identity.clone(), + pb::FindingChange { + kind: Kind::SeverityChanged as i32, + check_id: identity.suite.to_owned(), + before: Some(was[0].clone()), + after: Some(now[0].clone()), + occurrences: occurrences(shifted), + occurrences_before: total(&before_groups, identity), + occurrences_after: total(&after_groups, identity), + reason: String::new(), + origin: identity.origin.as_ref().map(Subject::entity_ref), + }, + )); + was.drain(..shifted); + now.drain(..shifted); + } + if !was.is_empty() { + gone.push(Pending { + identity: identity.clone(), + issues: was, + }); + } + if !now.is_empty() { + new.push(Pending { + identity: identity.clone(), + issues: now, + }); + } + } + + let mut predecessors: HashMap> = + HashMap::new(); + for identity in before_groups.keys() { + let (lineage, versions) = identity.lineage(); + predecessors + .entry(lineage) + .or_default() + .push((versions, identity)); + } + let mut gone_by_identity: HashMap = gone + .into_iter() + .map(|pending| (pending.identity.clone(), pending)) + .collect(); + + for pending in new { + let (lineage, versions) = pending.identity.lineage(); + let predecessor = predecessors.get(&lineage).and_then(|candidates| { + candidates + .iter() + .filter(|(v, _)| v.precedes(&versions)) + .max_by(|(a, _), (b, _)| a.cmp(b)) + .map(|(_, identity)| *identity) + }); + let identity = &pending.identity; + let change = if let Some(predecessor) = predecessor { + if let Some(left) = gone_by_identity.get_mut(predecessor) { + let carried = left.issues.len().min(pending.issues.len()); + left.issues.drain(..carried); + } + pb::FindingChange { + kind: Kind::CarriedToSuccessor as i32, + check_id: identity.suite.to_owned(), + before: before_groups + .get(predecessor) + .and_then(|issues| issues.first()) + .map(|issue| (*issue).clone()), + after: Some(pending.issues[0].clone()), + occurrences: occurrences(pending.issues.len()), + occurrences_before: total(&before_groups, predecessor), + occurrences_after: total(&after_groups, identity), + reason: String::new(), + origin: identity.origin.as_ref().map(Subject::entity_ref), + } + } else { + let gap = before.gap(identity.suite); + pb::FindingChange { + kind: if gap.is_some() { + Kind::Unknown + } else { + Kind::Introduced + } as i32, + check_id: identity.suite.to_owned(), + before: None, + after: Some(pending.issues[0].clone()), + occurrences: occurrences(pending.issues.len()), + occurrences_before: total(&before_groups, identity), + occurrences_after: total(&after_groups, identity), + reason: gap.unwrap_or_default(), + origin: identity.origin.as_ref().map(Subject::entity_ref), + } + }; + changes.push((identity.clone(), change)); + } + + for pending in gone_by_identity.into_values() { + let Some(first) = pending.issues.first() else { + continue; + }; + let identity = &pending.identity; + let removed = identity + .entities() + .any(|key| before.keys.contains(&key) && !after.keys.contains(&key)); + let gap = (!removed).then(|| after.gap(identity.suite)).flatten(); + let kind = match (removed, &gap) { + (true, _) => Kind::SubjectRemoved, + (false, Some(_)) => Kind::Unknown, + (false, None) => Kind::Resolved, + }; + changes.push(( + identity.clone(), + pb::FindingChange { + kind: kind as i32, + check_id: identity.suite.to_owned(), + before: Some((*first).clone()), + after: None, + occurrences: occurrences(pending.issues.len()), + occurrences_before: total(&before_groups, identity), + occurrences_after: total(&after_groups, identity), + reason: gap.unwrap_or_default(), + origin: identity.origin.as_ref().map(Subject::entity_ref), + }, + )); + } + + changes.sort_by(|(a, x), (b, y)| a.cmp(b).then(x.kind.cmp(&y.kind))); + let complete = !before.coverage.snapshot_truncated + && !after.coverage.snapshot_truncated + && !before.leaves_out_applicable_suites() + && !after.leaves_out_applicable_suites() + && COMPARED_SUITES + .iter() + .all(|suite| before.gap(suite).is_none() && after.gap(suite).is_none()); + pb::ConsequenceDelta { + changes: changes.into_iter().map(|(_, change)| change).collect(), + unchanged_count: occurrences(unchanged), + before_coverage: Some(before.coverage.clone()), + after_coverage: Some(after.coverage.clone()), + complete, + before_snapshot_id, + after_snapshot_id, + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use pb::{finding_change::Kind, validation_check_result::State, validation_issue::Severity}; + + use super::*; + + fn subject(slug: &str, version: u64) -> pb::EntityRef { + pb::EntityRef { + kind: pb::EntityKind::EventModel as i32, + id: Some(pb::Id { + namespace: "shop".into(), + slug: slug.into(), + version, + }), + } + } + + fn issue(code: &str, subject: pb::EntityRef, severity: Severity) -> pb::ValidationIssue { + pb::ValidationIssue { + severity: severity as i32, + code: code.into(), + message: format!("{code} prose"), + subject: Some(subject), + ..Default::default() + } + } + + fn scan( + findings: Vec, + keys: &[pb::EntityRef], + state: impl Fn(&str) -> (State, &'static str), + ) -> Scan { + Scan { + findings: findings + .into_iter() + .map(|issue| Finding { + suite: validation::CHECK_MODEL, + origin: None, + issue, + }) + .collect(), + coverage: pb::ValidationCoverage { + checks: COMPARED_SUITES + .iter() + .map(|suite| { + let (state, reason) = state(suite); + validation::coverage_check(*suite, state, reason, 1) + }) + .collect(), + ..Default::default() + }, + keys: keys.iter().filter_map(|r| Subject::of(r).key()).collect(), + } + } + + fn complete(_: &str) -> (State, &'static str) { + (State::Completed, "") + } + + fn delta(before: &Scan, after: &Scan) -> pb::ConsequenceDelta { + compare(before, after, "before".into(), "after".into()) + } + + fn kinds(delta: &pb::ConsequenceDelta) -> Vec<(i32, u32)> { + delta + .changes + .iter() + .map(|c| (c.kind, c.occurrences)) + .collect() + } + + #[test] + fn identical_scans_report_no_change() { + let m = subject("checkout", 1); + let findings = vec![issue("DANGLING_REF", m.clone(), Severity::Error)]; + let before = scan(findings.clone(), std::slice::from_ref(&m), complete); + let after = scan(findings, &[m], complete); + let delta = delta(&before, &after); + assert!(delta.changes.is_empty(), "{delta:?}"); + assert_eq!(delta.unchanged_count, 1); + assert!(delta.complete); + } + + #[test] + fn prose_does_not_change_identity() { + let m = subject("checkout", 1); + let mut reworded = issue("DANGLING_REF", m.clone(), Severity::Error); + reworded.message = "different words".into(); + let before = scan( + vec![issue("DANGLING_REF", m.clone(), Severity::Error)], + std::slice::from_ref(&m), + complete, + ); + let after = scan(vec![reworded], &[m], complete); + assert_eq!(delta(&before, &after).changes, Vec::new()); + } + + #[test] + fn introduced_and_resolved_need_complete_scans() { + let m = subject("checkout", 1); + let before = scan( + vec![issue("A", m.clone(), Severity::Error)], + std::slice::from_ref(&m), + complete, + ); + let after = scan( + vec![issue("B", m.clone(), Severity::Error)], + std::slice::from_ref(&m), + complete, + ); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::Resolved as i32, 1), (Kind::Introduced as i32, 1)] + ); + } + + #[test] + fn a_finding_is_not_resolved_when_the_after_scan_skipped_its_suite() { + let m = subject("checkout", 1); + let before = scan( + vec![issue("A", m.clone(), Severity::Error)], + std::slice::from_ref(&m), + complete, + ); + let after = scan(Vec::new(), &[m], |suite| { + if suite == validation::CHECK_MODEL { + (State::Skipped, "write_path") + } else { + (State::Completed, "") + } + }); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::Unknown as i32, 1)]); + assert_eq!(delta.changes[0].reason, "write_path"); + assert!(delta.changes[0].before.is_some() && delta.changes[0].after.is_none()); + assert!(!delta.complete); + } + + #[test] + fn a_finding_is_not_resolved_when_the_after_scan_omitted_its_suite() { + let m = subject("checkout", 1); + let before = scan( + vec![issue("A", m.clone(), Severity::Error)], + std::slice::from_ref(&m), + complete, + ); + let mut after = scan(Vec::new(), &[m], complete); + after + .coverage + .checks + .retain(|c| c.check_id != validation::CHECK_MODEL); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::Unknown as i32, 1)]); + assert_eq!(delta.changes[0].reason, REASON_SUITE_NOT_REPORTED); + } + + #[test] + fn a_finding_is_not_new_when_the_before_scan_was_partial() { + let m = subject("checkout", 1); + let before = scan(Vec::new(), std::slice::from_ref(&m), |_| { + (State::Partial, "snapshot_truncated") + }); + let after = scan(vec![issue("A", m.clone(), Severity::Error)], &[m], complete); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::Unknown as i32, 1)]); + assert_eq!(delta.changes[0].reason, "snapshot_truncated"); + assert!(delta.changes[0].after.is_some() && delta.changes[0].before.is_none()); + } + + #[test] + fn truncation_makes_the_delta_incomplete() { + let m = subject("checkout", 1); + let before = scan(Vec::new(), std::slice::from_ref(&m), complete); + let mut after = scan(Vec::new(), &[m], complete); + after.coverage.snapshot_truncated = true; + assert!(!delta(&before, &after).complete); + } + + #[test] + fn an_applicable_suite_the_delta_cannot_run_makes_it_incomplete() { + let m = subject("checkout", 1); + let before = scan(Vec::new(), std::slice::from_ref(&m), complete); + let mut after = scan(Vec::new(), &[m], complete); + after.coverage.checks.push(validation::coverage_check( + CHECK_SPECIFICATION_TYPES, + State::Skipped, + REASON_DELTA_NOT_SUPPORTED, + 0, + )); + assert!(!delta(&before, &after).complete); + } + + #[test] + fn severity_changes_are_reported_once_per_identity() { + let m = subject("checkout", 1); + let before = scan( + vec![issue("A", m.clone(), Severity::Warning)], + std::slice::from_ref(&m), + complete, + ); + let after = scan(vec![issue("A", m.clone(), Severity::Error)], &[m], complete); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::SeverityChanged as i32, 1)]); + assert_eq!( + delta.changes[0].before.as_ref().unwrap().severity, + Severity::Warning as i32 + ); + assert_eq!( + delta.changes[0].after.as_ref().unwrap().severity, + Severity::Error as i32 + ); + } + + #[test] + fn repeated_occurrences_are_counted() { + let m = subject("checkout", 1); + let one = issue("A", m.clone(), Severity::Warning); + let before = scan( + vec![one.clone(), one.clone()], + std::slice::from_ref(&m), + complete, + ); + let after = scan(vec![one.clone(), one.clone(), one], &[m], complete); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::Introduced as i32, 1)]); + assert_eq!( + ( + delta.changes[0].occurrences_before, + delta.changes[0].occurrences_after + ), + (2, 3) + ); + assert_eq!(delta.unchanged_count, 2); + } + + #[test] + fn findings_on_a_removed_subject_go_with_it() { + let m = subject("checkout", 1); + let before = scan(vec![issue("A", m.clone(), Severity::Error)], &[m], complete); + let after = scan(Vec::new(), &[], complete); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::SubjectRemoved as i32, 1)] + ); + } + + #[test] + fn a_successor_version_carries_its_predecessors_finding() { + let v1 = subject("checkout", 1); + let v2 = subject("checkout", 2); + let before = scan( + vec![issue("A", v1.clone(), Severity::Error)], + std::slice::from_ref(&v1), + complete, + ); + let after = scan( + vec![ + issue("A", v1.clone(), Severity::Error), + issue("A", v2.clone(), Severity::Error), + ], + &[v1, v2], + complete, + ); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::CarriedToSuccessor as i32, 1)]); + assert_eq!(delta.unchanged_count, 1); + } + + #[test] + fn a_successor_replacing_its_predecessor_consumes_the_old_finding() { + let v1 = subject("checkout", 1); + let v2 = subject("checkout", 2); + let before = scan( + vec![issue("A", v1.clone(), Severity::Error)], + &[v1], + complete, + ); + let after = scan( + vec![issue("A", v2.clone(), Severity::Error)], + &[v2], + complete, + ); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::CarriedToSuccessor as i32, 1)] + ); + } + + fn citing( + mut issue: pb::ValidationIssue, + entity: pb::EntityRef, + detail: &str, + ) -> pb::ValidationIssue { + issue.evidence.push(pb::ValidationEvidence { + role: pb::validation_evidence::Role::Issuer as i32, + entity: Some(entity), + field: "slices[0]".into(), + detail: detail.into(), + ..Default::default() + }); + issue + } + + #[test] + fn evidence_pinned_to_the_subject_is_carried_to_its_successor() { + let v1 = subject("checkout", 1); + let v2 = subject("checkout", 2); + let before = scan( + vec![citing( + issue("A", v1.clone(), Severity::Error), + v1.clone(), + "", + )], + &[v1], + complete, + ); + let after = scan( + vec![citing( + issue("A", v2.clone(), Severity::Error), + v2.clone(), + "", + )], + &[v2], + complete, + ); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::CarriedToSuccessor as i32, 1)] + ); + } + + #[test] + fn evidence_pinned_to_an_older_version_is_not_a_successor() { + let v1 = subject("checkout", 1); + let v2 = subject("checkout", 2); + let before = scan( + vec![citing( + issue("A", v1.clone(), Severity::Error), + v2.clone(), + "", + )], + &[v1.clone(), v2.clone()], + complete, + ); + let after = scan( + vec![citing( + issue("A", v2.clone(), Severity::Error), + v1.clone(), + "", + )], + &[v1, v2], + complete, + ); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::Resolved as i32, 1), (Kind::Introduced as i32, 1)] + ); + } + + #[test] + fn evidence_detail_does_not_change_identity() { + let m = subject("checkout", 1); + let before = scan( + vec![citing( + issue("A", m.clone(), Severity::Error), + m.clone(), + "first words", + )], + std::slice::from_ref(&m), + complete, + ); + let after = scan( + vec![citing( + issue("A", m.clone(), Severity::Error), + m.clone(), + "other words", + )], + &[m], + complete, + ); + assert_eq!(delta(&before, &after).changes, Vec::new()); + } + + fn originating(mut scan: Scan, origins: &[pb::EntityRef]) -> Scan { + for (finding, origin) in scan.findings.iter_mut().zip(origins) { + finding.origin = Some(Subject::of(origin)); + } + scan + } + + #[test] + fn one_rule_on_one_subject_from_two_origins_stays_two_findings() { + let event = subject("placed", 1); + let (a, b) = (subject("a", 1), subject("b", 1)); + let before = originating( + scan( + vec![issue("A", event.clone(), Severity::Error)], + &[a.clone(), b.clone()], + complete, + ), + std::slice::from_ref(&a), + ); + let after = originating( + scan( + vec![ + issue("A", event.clone(), Severity::Error), + issue("A", event, Severity::Error), + ], + &[a.clone(), b.clone()], + complete, + ), + &[a, b.clone()], + ); + let delta = delta(&before, &after); + assert_eq!(kinds(&delta), vec![(Kind::Introduced as i32, 1)]); + assert_eq!(delta.changes[0].origin, Some(b)); + } + + #[test] + fn a_finding_whose_origin_is_removed_is_not_resolved() { + let event = subject("placed", 1); + let model = subject("checkout", 1); + let before = originating( + scan( + vec![issue("A", event.clone(), Severity::Error)], + &[event.clone(), model.clone()], + complete, + ), + &[model], + ); + let after = scan(vec![], &[event], complete); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::SubjectRemoved as i32, 1)] + ); + } + + #[test] + fn a_finding_on_a_newer_origin_is_carried() { + let event = subject("placed", 1); + let (v1, v2) = (subject("checkout", 1), subject("checkout", 2)); + let before = originating( + scan( + vec![issue("A", event.clone(), Severity::Error)], + &[event.clone(), v1.clone()], + complete, + ), + &[v1], + ); + let after = originating( + scan( + vec![issue("A", event.clone(), Severity::Error)], + &[event, v2.clone()], + complete, + ), + &[v2], + ); + assert_eq!( + kinds(&delta(&before, &after)), + vec![(Kind::CarriedToSuccessor as i32, 1)] + ); + } + + #[test] + fn structured_issue_fields_beyond_the_known_ones_join_identity() { + let m = subject("checkout", 1); + let identity = FindingIdentity::of(&Finding { + suite: validation::CHECK_MODEL, + origin: None, + issue: issue("A", m.clone(), Severity::Error), + }); + assert_eq!(identity.structure, Vec::::new()); + let mut other = identity.clone(); + other.structure = vec![1]; + assert_ne!(identity, other); + } +} diff --git a/rsworkspace/crates/trogon-atlas-server/src/lib.rs b/rsworkspace/crates/trogon-atlas-server/src/lib.rs index 11d983bb..eed65df6 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/lib.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/lib.rs @@ -4,6 +4,7 @@ pub mod auth; pub mod batch_recovery; pub mod change_cursor; pub mod changeset_receipt; +pub mod consequence; pub mod conv; mod data_product_resources; mod decision_document_resources; diff --git a/rsworkspace/crates/trogon-atlas-server/src/metrics.rs b/rsworkspace/crates/trogon-atlas-server/src/metrics.rs index 07a6593d..8369324e 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/metrics.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/metrics.rs @@ -17,6 +17,7 @@ pub struct ServerMetrics { rpc_server_call_duration: metric::RpcServerCallDuration, rpc_server_stream_terminations: metric::RpcServerStreamTerminations, mutation_lock_wait_duration: metric::MutationLockWaitDuration, + consequence_delta_duration: metric::ConsequenceDeltaDuration, cache_lookups: metric::CacheLookups, change_stream_message_count: metric::ChangeStreamMessageCount, change_stream_size: metric::ChangeStreamSize, @@ -42,6 +43,7 @@ impl ServerMetrics { rpc_server_call_duration: metric::RpcServerCallDuration::new(meter), rpc_server_stream_terminations: metric::RpcServerStreamTerminations::new(meter), mutation_lock_wait_duration: metric::MutationLockWaitDuration::new(meter), + consequence_delta_duration: metric::ConsequenceDeltaDuration::new(meter), cache_lookups: metric::CacheLookups::new(meter), change_stream_message_count: metric::ChangeStreamMessageCount::new(meter), change_stream_size: metric::ChangeStreamSize::new(meter), @@ -86,6 +88,12 @@ impl ServerMetrics { .record(waited.as_secs_f64()); } + /// Time spent computing one consequence delta. + pub fn record_consequence_delta(&self, elapsed: Duration) { + self.consequence_delta_duration + .record(elapsed.as_secs_f64()); + } + pub fn record_snapshot_cache_hit(&self) { self.record_cache_lookup(attribute::CacheName::Snapshot, attribute::CacheResult::Hit); } diff --git a/rsworkspace/crates/trogon-atlas-server/src/service.rs b/rsworkspace/crates/trogon-atlas-server/src/service.rs index 042ad5c1..aa1ac1dd 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/service.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/service.rs @@ -57,7 +57,7 @@ use trogon_atlas_store::{ }; use crate::{ - analysis, as_of, + analysis, as_of, consequence, conv::{ entity_id, entity_kind, kind_from_i32, require_id, store_err, validate_branch_name, validate_entity_content, @@ -1125,6 +1125,65 @@ impl EventModelServiceImpl { Ok((entities, result.truncated)) } + /// How replacing `before` with `after` moves the findings of every + /// compared suite. `truncated` says the `before` view hit the entity + /// cap; `after` is derived from it, so it inherits the gap. + /// Advisory: a failure to compute the delta never fails the dry run it + /// rides on, so this reports `None` instead of an error. + async fn consequence_delta( + &self, + before: &[StoredEntity], + after: &[StoredEntity], + truncated: bool, + branch: Option<&str>, + vis: &Visibility, + ) -> Option { + let started = std::time::Instant::now(); + let contexts = async { + Ok::<_, Status>(( + self.architecture_validation_context(before, branch, vis) + .await?, + self.architecture_validation_context(after, branch, vis) + .await?, + )) + }; + let (before_context, after_context) = match contexts.await { + Ok(contexts) => contexts, + Err(status) => { + tracing::warn!(error = %status, "consequence delta skipped"); + return None; + } + }; + let before_context = consequence::context_outside(before_context, &[before]); + let after_context = consequence::context_outside(after_context, &[before, after]); + let scope = pb::ValidationScope { + branch: branch.unwrap_or_default().to_owned(), + ..Default::default() + }; + let all_namespaces = std::collections::BTreeSet::new(); + let snapshot_id = |world: &[StoredEntity]| { + snapshot_digest(world, &all_namespaces, false).map(|digest| digest.snapshot_id) + }; + let mut delta = consequence::compare( + &consequence::Scan::of(before, &before_context, truncated, &scope), + &consequence::Scan::of(after, &after_context, truncated, &scope), + String::new(), + String::new(), + ); + match (snapshot_id(before), snapshot_id(after)) { + (Ok(before_id), Ok(after_id)) => { + delta.before_snapshot_id = before_id; + delta.after_snapshot_id = after_id; + } + (Err(status), _) | (_, Err(status)) => { + tracing::debug!(error = %status, "consequence delta has no snapshot binding"); + delta.complete = false; + } + } + self.metrics.record_consequence_delta(started.elapsed()); + Some(delta) + } + async fn lock_mutations(&self) -> tokio::sync::MutexGuard<'_, ()> { let started = std::time::Instant::now(); let guard = self.mutation_lock.lock().await; @@ -2878,6 +2937,7 @@ impl EventModelServiceImpl { precondition_failure: None, operation_receipt: None, coverage: None, + consequence_delta: None, } } @@ -2900,6 +2960,7 @@ impl EventModelServiceImpl { precondition_failure: None, operation_receipt: Some(operation_receipts::replay_receipt(operation_id, record)), coverage: None, + consequence_delta: None, })) } @@ -2926,6 +2987,7 @@ impl EventModelServiceImpl { validation: Vec::new(), operation_receipt: Some(operation_receipts::replay_receipt(operation_id, record)), coverage: None, + consequence_delta: None, })) } @@ -2991,6 +3053,7 @@ impl EventModelServiceImpl { precondition_failure: None, operation_receipt: None, coverage: None, + consequence_delta: None, } } @@ -3191,6 +3254,82 @@ fn world_after( world.into_iter().flatten().collect() } +/// `base` after the overlay `outcomes`, each paired with the key it landed +/// on. Etags are dropped: the overlay's are not the store's. +fn world_with_outcomes<'a>( + base: &[StoredEntity], + keys: impl IntoIterator, + outcomes: &[MutationOutcome], +) -> Vec { + let changes: Vec<_> = keys + .into_iter() + .zip(outcomes) + .filter_map(|((kind, id), outcome)| match outcome { + MutationOutcome::Wrote { entity, .. } => Some((kind, id, Some(entity))), + MutationOutcome::Deleted => Some((kind, id, None)), + MutationOutcome::Noop { .. } => None, + }) + .collect(); + world_after(base, &changes) +} + +struct SnapshotDigest { + snapshot_id: String, + entity_count: u32, + entries: Vec, +} + +/// Content identity of `all`, restricted to `scope` when it is not empty. +fn snapshot_digest( + all: &[StoredEntity], + scope: &std::collections::BTreeSet<&str>, + include_entries: bool, +) -> Result { + let mut hasher = trogon_atlas_core::SnapshotHasher::new(); + let mut entries = Vec::new(); + for se in all { + let id = entity_id(&se.entity)?; + if !scope.is_empty() && !scope.contains(id.namespace.as_str()) { + continue; + } + let kind = entity_kind(&se.entity)?; + let key = trogon_atlas_store::key::entity_key(kind, id); + let hash = trogon_atlas_core::entity_content_hash(&se.entity).map_err(|e| { + // An entity that cannot render as canonical JSON has no + // content identity, so the snapshot has none either. Failing + // is the honest answer: silently skipping it would produce an + // id for a model that is missing an entity. + Status::failed_precondition(format!( + "entity {key} has no content identity, so no snapshot id can be computed: {e}" + )) + })?; + if include_entries { + entries.push(pb::SnapshotEntry { + key: key.clone(), + entity: Some(pb::EntityRef { + kind: kind as i32, + id: Some(id.clone()), + }), + content_hash: hash.to_hex(), + }); + } + hasher.insert(key, hash).map_err(|e| { + // Two rows for one key means the store enumeration is + // broken; hashing on would name a state that never existed. + Status::internal(format!("snapshot enumeration produced {e}")) + })?; + } + // `SnapshotHasher` keeps its own sorted order; sort the response + // entries to match, since the proto promises key order. + entries.sort_by(|a, b| a.key.cmp(&b.key)); + + Ok(SnapshotDigest { + snapshot_id: hasher.finish().to_hex(), + entity_count: u32::try_from(hasher.len()).unwrap_or(u32::MAX), + entries, + }) +} + fn runs_scenarios(kind: pb::EntityKind) -> bool { matches!( kind, @@ -3249,6 +3388,7 @@ fn build_op_results( // Suite coverage for the batch lives on // BatchMutateResponse.coverage. coverage: None, + consequence_delta: None, }, )), } @@ -4060,49 +4200,12 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl let scope: std::collections::BTreeSet<&str> = req.namespaces.iter().map(String::as_str).collect(); - let mut hasher = trogon_atlas_core::SnapshotHasher::new(); - let mut entries = Vec::new(); - for se in all.iter() { - let id = entity_id(&se.entity)?; - if !scope.is_empty() && !scope.contains(id.namespace.as_str()) { - continue; - } - let kind = entity_kind(&se.entity)?; - let key = trogon_atlas_store::key::entity_key(kind, id); - let hash = trogon_atlas_core::entity_content_hash(&se.entity).map_err(|e| { - // An entity that cannot render as canonical JSON has no - // content identity, so the snapshot has none either. Failing - // is the honest answer: silently skipping it would produce an - // id for a model that is missing an entity. - Status::failed_precondition(format!( - "entity {key} has no content identity, so no snapshot id can be computed: {e}" - )) - })?; - if req.include_entries { - entries.push(pb::SnapshotEntry { - key: key.clone(), - entity: Some(pb::EntityRef { - kind: kind as i32, - id: Some(id.clone()), - }), - content_hash: hash.to_hex(), - }); - } - hasher.insert(key, hash).map_err(|e| { - // Two rows for one key means the store enumeration is - // broken; hashing on would name a state that never existed. - Status::internal(format!("snapshot enumeration produced {e}")) - })?; - } - // `SnapshotHasher` keeps its own sorted order; sort the response - // entries to match, since the proto promises key order. - entries.sort_by(|a, b| a.key.cmp(&b.key)); - + let digest = snapshot_digest(&all, &scope, req.include_entries)?; Ok(Response::new(pb::GetSnapshotIdResponse { - snapshot_id: hasher.finish().to_hex(), - entity_count: u32::try_from(hasher.len()).unwrap_or(u32::MAX), + snapshot_id: digest.snapshot_id, + entity_count: digest.entity_count, truncated, - entries, + entries: digest.entries, })) } @@ -6098,7 +6201,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl )?) }; - let _guard = self.lock_mutations().await; + let guard = self.lock_mutations().await; let settled = operation_receipts::settled(&*self.store, attempt.as_ref()).await; let resumed = match (&protect_reason, EntityIdentity::of(kind, &id)) { (Some(_), Some(identity)) => self @@ -6176,6 +6279,17 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl let prepared = trogon_atlas_core::connection_persistence::prepare_entity(previous, &entity) .map_err(|error| Status::invalid_argument(error.to_string()))?; + if branch.is_none() { + self.invalidate_snapshot().await; + } + let (before, truncated) = self + .snapshot_with_truncation(branch.as_deref(), &vis) + .await?; + let after = world_after(&before, &[(kind, &id, Some(&prepared))]); + drop(guard); + let consequence_delta = self + .consequence_delta(&before, &after, truncated, branch.as_deref(), &vis) + .await; return Ok(Response::new(pb::PutEntityResponse { entity: Some(prepared), etag: String::new(), @@ -6183,6 +6297,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl no_op: false, operation_receipt: None, coverage: Some(put_coverage), + consequence_delta, })); } @@ -6234,6 +6349,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl changeset_id, )), coverage: Some(put_coverage), + consequence_delta: None, })); } @@ -6451,6 +6567,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl operation_receipts::fresh_receipt(&req.operation_id, &receipt_changeset_id) }), coverage: Some(put_coverage), + consequence_delta: None, })) } @@ -7021,7 +7138,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl )?) }; - let _guard = self.lock_mutations().await; + let guard = self.lock_mutations().await; let settled = operation_receipts::settled(&*self.store, attempt.as_ref()).await; let resumed = self .protections_resumed_by( @@ -7264,6 +7381,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl subject_count(metas.len()), reached, )), + consequence_delta: None, })); } } @@ -7368,7 +7486,9 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl if branch.is_none() { self.invalidate_snapshot().await; } - let snapshot = self.snapshot_for(branch.as_deref(), &vis).await?; + let (snapshot, snapshot_truncated) = self + .snapshot_with_truncation(branch.as_deref(), &vis) + .await?; let mut exact: HashMap<_, _> = snapshot .iter() .filter_map(|stored| { @@ -7498,6 +7618,21 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl &per_op_referrers, referrers_incomplete, ); + let after = world_with_outcomes( + &pre_all_arc, + metas.iter().map(|meta| (meta.kind, &meta.id)), + &outcomes, + ); + drop(guard); + let consequence_delta = self + .consequence_delta( + &pre_all_arc, + &after, + snapshot_truncated, + branch.as_deref(), + &vis, + ) + .await; Ok(Response::new(pb::BatchMutateResponse { status: pb::batch_mutate_response::Status::Validated as i32, results, @@ -7506,6 +7641,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl precondition_failure: None, operation_receipt: None, coverage: Some(batch_coverage.clone()), + consequence_delta, })) } Err(err) => Ok(Response::new( @@ -7707,6 +7843,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl ) }), coverage: Some(batch_coverage), + consequence_delta: None, })) } // A partial apply leaves the claim deliberately unsettled: @@ -9561,7 +9698,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl return Err(branch_not_found(&req.name)); } - let _guard = self.lock_mutations().await; + let guard = self.lock_mutations().await; let deltas = self .store @@ -9740,61 +9877,28 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl // Full detection already ran above; validate the post-merge // world via an in-memory overlay without persisting anything. let (pre_all, snapshot_truncated) = self.snapshot_with_truncation(None, &vis).await?; - let post_world: Vec = match dry_run::apply_landing(&pre_all, &ops) { - Ok(outcomes) => { - let mut world: Vec = (*pre_all).clone(); - let mut by_key: std::collections::HashMap = - std::collections::HashMap::new(); - for (i, se) in world.iter().enumerate() { - if let (Some(k), Some(id)) = ( - trogon_atlas_store::refs::entity_kind(&se.entity), - trogon_atlas_store::refs::entity_id(&se.entity), - ) { - by_key.insert(pb::EntityKey::new(k, id), i); - } - } - for (meta, oc) in op_metas.iter().zip(outcomes.iter()) { - let key = pb::EntityKey::new(meta.kind, &meta.id); - match oc { - MutationOutcome::Wrote { entity, etag } => { - let se = StoredEntity { - entity: entity.clone(), - etag: etag.clone(), - }; - if let Some(&i) = by_key.get(&key) { - world[i] = se; - } else { - by_key.insert(key, world.len()); - world.push(se); - } - } - MutationOutcome::Noop { .. } => {} - MutationOutcome::Deleted => { - if let Some(&i) = by_key.get(&key) { - world.remove(i); - by_key.remove(&key); - for v in by_key.values_mut() { - if *v > i { - *v -= 1; - } - } - } - } - } - } - world - } + let post_world = match dry_run::apply_landing(&pre_all, &ops) { + Ok(outcomes) => world_with_outcomes( + &pre_all, + op_metas.iter().map(|meta| (meta.kind, &meta.id)), + &outcomes, + ), Err(error) => return Err(store_err(error)), }; let mut validation_issues = validation::validate_store_hygiene(&post_world); let has_error = validation_issues .iter() .any(|i| i.severity == pb::validation_issue::Severity::Error as i32); + drop(guard); + let consequence_delta = self + .consequence_delta(&pre_all, &post_world, snapshot_truncated, None, &vis) + .await; if has_error { return Ok(Response::new(pb::MergeBranchResponse { status: pb::merge_branch_response::Status::Invalid as i32, validation: validation_issues, coverage: Some(merge_coverage(snapshot_truncated, false)), + consequence_delta, ..landed.into_response() })); } @@ -9803,6 +9907,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl status: pb::merge_branch_response::Status::Validated as i32, validation: validation_issues, coverage: Some(merge_coverage(snapshot_truncated, true)), + consequence_delta, ..landed.into_response() })); } diff --git a/rsworkspace/crates/trogon-atlas-server/src/validation.rs b/rsworkspace/crates/trogon-atlas-server/src/validation.rs index 5a3205ac..79452d0d 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/validation.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/validation.rs @@ -1321,6 +1321,17 @@ pub fn validate_store(all: &[StoredEntity]) -> Vec { pub fn validate_store_with_architecture_context( all: &[StoredEntity], architecture_context: &[StoredEntity], +) -> Vec { + let mut issues = validate_architecture_resources(all, architecture_context); + issues.extend(validate_store_hygiene(all)); + issues +} + +/// Architecture resource rules and membership advisories for every resource in +/// `all`, resolved against `all` plus `architecture_context`. +pub(crate) fn validate_architecture_resources( + all: &[StoredEntity], + architecture_context: &[StoredEntity], ) -> Vec { let mut issues: Vec = Vec::new(); let mut resources = all @@ -1339,7 +1350,6 @@ pub fn validate_store_with_architecture_context( } } issues.extend(membership_advisories(all, architecture_context)); - issues.extend(validate_store_hygiene(all)); issues } diff --git a/rsworkspace/crates/trogon-atlas-server/tests/business_data.rs b/rsworkspace/crates/trogon-atlas-server/tests/business_data.rs index 82e0365f..28726b6e 100644 --- a/rsworkspace/crates/trogon-atlas-server/tests/business_data.rs +++ b/rsworkspace/crates/trogon-atlas-server/tests/business_data.rs @@ -3799,7 +3799,7 @@ async fn prepare_order_fixture(client: &mut BusinessClient) -> OrderFixture { #[tokio::test(flavor = "multi_thread")] async fn authenticated_order_fixture_keeps_business_declarations_explicit() { let mut client = authenticated_business_server().await; - let mut fixture = prepare_order_fixture(&mut client).await; + let mut fixture = Box::pin(prepare_order_fixture(&mut client)).await; let mut flagged = fixture.product.entity.clone(); match &mut flagged.kind { diff --git a/rsworkspace/crates/trogon-atlas-server/tests/consequence_delta.rs b/rsworkspace/crates/trogon-atlas-server/tests/consequence_delta.rs new file mode 100644 index 00000000..a5e77c76 --- /dev/null +++ b/rsworkspace/crates/trogon-atlas-server/tests/consequence_delta.rs @@ -0,0 +1,437 @@ +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::sync::Arc; + +use pb::{event_model_service_server::EventModelService, finding_change::Kind}; +use tonic::Request; +use trogon_atlas_proto as pb; +use trogon_atlas_server::{metrics::ServerMetrics, service::EventModelServiceImpl}; + +const BRANCH: &str = "consequence"; + +async fn service() -> EventModelServiceImpl { + let nats = trogon_atlas_testsupport::shared().await; + EventModelServiceImpl::try_new(Arc::new(nats.store().await), ServerMetrics::default()).unwrap() +} + +fn id(namespace: &str, slug: &str) -> pb::Id { + pb::Id { + namespace: namespace.into(), + slug: slug.into(), + version: 1, + } +} + +fn event(namespace: &str, slug: &str) -> pb::Entity { + pb::Entity { + kind: Some(pb::entity::Kind::Event(pb::Event { + id: Some(id(namespace, slug)), + title: format!("{namespace}/{slug}"), + ..Default::default() + })), + ..Default::default() + } +} + +fn event_model(namespace: &str, slug: &str, members: &[(pb::EntityKind, pb::Id)]) -> pb::Entity { + pb::Entity { + kind: Some(pb::entity::Kind::EventModel(pb::EventModel { + id: Some(id(namespace, slug)), + title: format!("{namespace}/{slug}"), + members: members + .iter() + .map(|(kind, id)| pb::EntityRef { + kind: *kind as i32, + id: Some(id.clone()), + }) + .collect(), + ..Default::default() + })), + ..Default::default() + } +} + +fn member_event(namespace: &str, slug: &str) -> (pb::EntityKind, pb::Id) { + (pb::EntityKind::Event, id(namespace, slug)) +} + +fn put_request(entity: pb::Entity, validate_only: bool) -> pb::PutEntityRequest { + pb::PutEntityRequest { + entity: Some(entity), + validate_only, + ..Default::default() + } +} + +async fn put(service: &EventModelServiceImpl, entity: pb::Entity) { + service + .put_entity(Request::new(put_request(entity, false))) + .await + .unwrap(); +} + +async fn preview_put(service: &EventModelServiceImpl, entity: pb::Entity) -> pb::ConsequenceDelta { + service + .put_entity(Request::new(put_request(entity, true))) + .await + .unwrap() + .into_inner() + .consequence_delta + .expect("a dry run reports its consequence delta") +} + +fn on_branch(value: T) -> Request { + let mut request = Request::new(value); + request + .metadata_mut() + .insert("x-trogon-atlas-branch", BRANCH.parse().unwrap()); + request +} + +fn slug(entity: Option<&pb::EntityRef>) -> &str { + entity + .and_then(|e| e.id.as_ref()) + .map_or("", |id| id.slug.as_str()) +} + +/// Each change as (kind, rule code, subject slug, origin slug). +fn changes(delta: &pb::ConsequenceDelta) -> Vec<(Kind, &str, &str, &str)> { + delta + .changes + .iter() + .map(|change| { + let issue = change.after.as_ref().or(change.before.as_ref()).unwrap(); + ( + Kind::try_from(change.kind).unwrap(), + issue.code.as_str(), + slug(issue.subject.as_ref()), + slug(change.origin.as_ref()), + ) + }) + .collect() +} + +fn kinds_of<'a>(delta: &'a pb::ConsequenceDelta, code: &str) -> Vec<(Kind, &'a str, &'a str)> { + changes(delta) + .into_iter() + .filter(|(_, c, _, _)| *c == code) + .map(|(kind, _, subject, origin)| (kind, subject, origin)) + .collect() +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unchanged_world_reports_no_change() { + let service = service().await; + put(&service, event("cd-noop", "placed")).await; + let model = event_model("cd-noop", "checkout", &[member_event("cd-noop", "placed")]); + put(&service, model.clone()).await; + + let delta = preview_put(&service, model).await; + + assert!(delta.changes.is_empty(), "{delta:?}"); + assert!(delta.complete, "{delta:?}"); + assert_eq!(delta.before_snapshot_id, delta.after_snapshot_id); + assert_ne!(delta.before_snapshot_id, ""); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_new_error_is_introduced() { + let service = service().await; + put(&service, event("cd-new", "placed")).await; + put( + &service, + event_model("cd-new", "checkout", &[member_event("cd-new", "placed")]), + ) + .await; + + let delta = preview_put( + &service, + event_model( + "cd-new", + "checkout", + &[ + member_event("cd-new", "placed"), + member_event("cd-new", "missing"), + ], + ), + ) + .await; + + assert_eq!( + kinds_of(&delta, "MEMBER_NOT_FOUND"), + vec![(Kind::Introduced, "missing", "checkout")], + "{delta:?}" + ); + let change = delta + .changes + .iter() + .find(|c| { + c.after + .as_ref() + .is_some_and(|i| i.code == "MEMBER_NOT_FOUND") + }) + .unwrap(); + assert_eq!(change.check_id, "model"); + assert!(delta.complete); + assert_ne!(delta.before_snapshot_id, delta.after_snapshot_id); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_fixed_error_is_resolved_and_unrelated_findings_stay_unchanged() { + let service = service().await; + put( + &service, + event_model("cd-fix", "checkout", &[member_event("cd-fix", "missing")]), + ) + .await; + put( + &service, + event_model("cd-fix", "refunds", &[member_event("cd-fix", "elsewhere")]), + ) + .await; + + let delta = preview_put(&service, event("cd-fix", "missing")).await; + + assert_eq!( + kinds_of(&delta, "MEMBER_NOT_FOUND"), + vec![(Kind::Resolved, "missing", "checkout")], + "{delta:?}" + ); + assert!( + changes(&delta) + .iter() + .all(|(_, _, subject, origin)| { *subject != "elsewhere" && *origin != "refunds" }), + "the untouched model's findings stay unchanged: {delta:?}" + ); + assert!(delta.unchanged_count >= 1, "{delta:?}"); + assert!(delta.complete); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_batch_delete_reports_the_finding_it_causes_on_another_model() { + let service = service().await; + put(&service, event("cd-cross", "placed")).await; + put( + &service, + event_model( + "cd-cross", + "checkout", + &[member_event("cd-cross", "placed")], + ), + ) + .await; + + let response = service + .batch_mutate(Request::new(pb::BatchMutateRequest { + ops: vec![pb::BatchMutateOp { + op: Some(pb::batch_mutate_op::Op::Delete(pb::DeleteEntityRequest { + kind: pb::EntityKind::Event as i32, + id: Some(id("cd-cross", "placed")), + mode: pb::delete_entity_request::Mode::Force as i32, + ..Default::default() + })), + }], + validate_only: true, + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + + assert_eq!( + response.status, + pb::batch_mutate_response::Status::Validated as i32, + "{response:?}" + ); + let delta = response.consequence_delta.expect("dry run reports a delta"); + assert_eq!( + kinds_of(&delta, "MEMBER_NOT_FOUND"), + vec![(Kind::Introduced, "placed", "checkout")], + "{delta:?}" + ); + service + .get_entity(Request::new(pb::GetEntityRequest { + kind: pb::EntityKind::Event as i32, + id: Some(id("cd-cross", "placed")), + })) + .await + .expect("a dry run never deletes"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn findings_of_a_deleted_model_leave_with_it() { + let service = service().await; + put( + &service, + event_model("cd-gone", "checkout", &[member_event("cd-gone", "missing")]), + ) + .await; + + let delta = service + .batch_mutate(Request::new(pb::BatchMutateRequest { + ops: vec![pb::BatchMutateOp { + op: Some(pb::batch_mutate_op::Op::Delete(pb::DeleteEntityRequest { + kind: pb::EntityKind::EventModel as i32, + id: Some(id("cd-gone", "checkout")), + mode: pb::delete_entity_request::Mode::Force as i32, + ..Default::default() + })), + }], + validate_only: true, + ..Default::default() + })) + .await + .unwrap() + .into_inner() + .consequence_delta + .expect("dry run reports a delta"); + + assert_eq!( + kinds_of(&delta, "MEMBER_NOT_FOUND"), + vec![(Kind::SubjectRemoved, "missing", "checkout")], + "{delta:?}" + ); + assert!( + changes(&delta) + .iter() + .all(|(kind, _, _, _)| *kind != Kind::Resolved), + "deleting a model fixes nothing: {delta:?}" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn models_that_reference_each_other_terminate() { + let service = service().await; + let a = (pb::EntityKind::EventModel, id("cd-cycle", "a")); + let b = (pb::EntityKind::EventModel, id("cd-cycle", "b")); + put( + &service, + event_model("cd-cycle", "a", std::slice::from_ref(&b)), + ) + .await; + put( + &service, + event_model("cd-cycle", "b", std::slice::from_ref(&a)), + ) + .await; + + let delta = preview_put( + &service, + event_model("cd-cycle", "a", &[b, member_event("cd-cycle", "missing")]), + ) + .await; + + assert!( + changes(&delta) + .iter() + .any(|(kind, _, subject, origin)| *kind == Kind::Introduced + && (*subject == "missing" || *subject == "a" || *origin == "a")), + "{delta:?}" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_truncated_scan_never_claims_a_resolution() { + let service = service().await.with_snapshot_cap(1); + put( + &service, + event_model( + "cd-trunc", + "checkout", + &[member_event("cd-trunc", "missing")], + ), + ) + .await; + put(&service, event("cd-trunc", "other")).await; + + let delta = preview_put(&service, event("cd-trunc", "missing")).await; + + assert!(!delta.complete, "{delta:?}"); + let after = delta.after_coverage.as_ref().unwrap(); + assert!(after.snapshot_truncated, "{after:?}"); + assert!( + delta + .changes + .iter() + .all(|change| change.kind == Kind::Unknown as i32), + "a partial scan must not classify: {delta:?}" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_merge_preview_reports_what_landing_resolves() { + let service = service().await; + put( + &service, + event_model( + "cd-merge", + "checkout", + &[member_event("cd-merge", "missing")], + ), + ) + .await; + service + .create_branch(Request::new(pb::CreateBranchRequest { + name: BRANCH.into(), + ..Default::default() + })) + .await + .unwrap(); + service + .put_entity(on_branch(put_request(event("cd-merge", "missing"), false))) + .await + .unwrap(); + + let response = service + .merge_branch(Request::new(pb::MergeBranchRequest { + name: BRANCH.into(), + dry_run: true, + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + + let delta = response + .consequence_delta + .expect("merge preview reports a delta"); + assert_eq!( + kinds_of(&delta, "MEMBER_NOT_FOUND"), + vec![(Kind::Resolved, "missing", "checkout")], + "{delta:?}" + ); + assert!(delta.complete); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_real_write_carries_no_delta() { + let service = service().await; + let response = service + .put_entity(Request::new(put_request( + event("cd-write", "placed"), + false, + ))) + .await + .unwrap() + .into_inner(); + assert!(response.consequence_delta.is_none()); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_world_without_content_identity_still_gets_an_unbound_delta() { + let service = service().await; + let mut opaque = event("cd-opaque", "priced"); + if let Some(pb::entity::Kind::Event(event)) = opaque.kind.as_mut() { + event.metadata.push(prost_types::Any { + type_url: "type.googleapis.com/acme.unknown.v1.Thing".into(), + value: vec![8, 1], + }); + } + put(&service, opaque).await; + + let delta = preview_put(&service, event("cd-opaque", "other")).await; + + assert!(!delta.complete, "{delta:?}"); + assert_eq!(delta.before_snapshot_id, ""); + assert_eq!(delta.after_snapshot_id, ""); +} diff --git a/rsworkspace/crates/trogon-atlas-server/tests/design_topology_branch_backend.rs b/rsworkspace/crates/trogon-atlas-server/tests/design_topology_branch_backend.rs index 82688135..945a4778 100644 --- a/rsworkspace/crates/trogon-atlas-server/tests/design_topology_branch_backend.rs +++ b/rsworkspace/crates/trogon-atlas-server/tests/design_topology_branch_backend.rs @@ -360,10 +360,10 @@ async fn restoration_renews_only_restored_identities(take_theirs: bool) { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn keep_ours_restoration_renews_only_the_restored_step_and_transitions() { - restoration_renews_only_restored_identities(false).await; + Box::pin(restoration_renews_only_restored_identities(false)).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn take_theirs_restoration_renews_only_the_restored_step_and_transitions() { - restoration_renews_only_restored_identities(true).await; + Box::pin(restoration_renews_only_restored_identities(true)).await; } diff --git a/rsworkspace/crates/trogon-atlas-server/tests/message_claims_acceptance.rs b/rsworkspace/crates/trogon-atlas-server/tests/message_claims_acceptance.rs index 60ed7841..d5c12676 100644 --- a/rsworkspace/crates/trogon-atlas-server/tests/message_claims_acceptance.rs +++ b/rsworkspace/crates/trogon-atlas-server/tests/message_claims_acceptance.rs @@ -2553,7 +2553,7 @@ async fn standalone_native_schema_hidden_and_missing_are_identical() { .namespace .unwrap() .id; - let (leaf, foreign_id) = foreign_native_schema_leaf(&mut client, &namespace).await; + let (leaf, foreign_id) = Box::pin(foreign_native_schema_leaf(&mut client, &namespace)).await; let mut hidden = Vec::new(); for kind in [ pb::EntityKind::Document, @@ -2665,7 +2665,7 @@ async fn standalone_native_schema_authorization_and_decode_faults_remain_errors( .namespace .unwrap() .id; - let (leaf, foreign_id) = foreign_native_schema_leaf(&mut client, &namespace).await; + let (leaf, foreign_id) = Box::pin(foreign_native_schema_leaf(&mut client, &namespace)).await; *fault.target.lock().unwrap() = Some(foreign_id.namespace.clone()); for kind in [ pb::EntityKind::Document,