diff --git a/.fluree-memory/repo.ttl b/.fluree-memory/repo.ttl index 3762c03da..fd95869a9 100644 --- a/.fluree-memory/repo.ttl +++ b/.fluree-memory/repo.ttl @@ -1461,6 +1461,33 @@ mem:fact-01kj8k5nzpe59p5vp7zrehbrr9 a mem:Fact ; mem:branch "feature/memory" ; mem:createdAt "2026-02-24T19:49:14.358637+00:00"^^xsd:dateTime . +mem:decision-01kwsmdnf7mhnbvn29qgf2j2cx a mem:Decision ; + mem:content "Merge-preview include_changes nets per FULL fact key (g,s,p,o,dt,lang,i) — NOT Flake Eq/Hash (ignores g) and NOT (s,p,o,dt,g) alone (collides lang tags + list positions). Netting contract = \"net commit effect\": fact survives iff oldest and newest in-range ops agree (net op = newest); re-assert of a pre-range value nets as assert because the merge replays it. Changes are strategy-independent (raw source-vs-ancestor delta, pre-conflict-resolution) — under TakeSource/TakeBranch the merge commit differs on conflict keys; conflicts.details covers those." ; + mem:tag "branches" ; + mem:tag "diff" ; + mem:tag "flake-identity" ; + mem:tag "merge-preview" ; + mem:tag "netting" ; + mem:scope mem:repo ; + mem:artifactRef "fluree-db-api/src/merge_preview.rs" ; + mem:artifactRef "fluree-db-novelty/src/delta.rs" ; + mem:branch "feature/merge-preview-changes" ; + mem:createdAt "2026-07-05T17:15:08.135861+00:00"^^xsd:dateTime ; + mem:rationale "Pairing-only netting cannot distinguish ancestor-state diff from commit-effect diff without loading ancestor state; net-commit-effect is cheap and equals what merge_branch applies minus cancelling pairs." . + +mem:fact-01kwsmdy5pms38yy81d0bdxprb a mem:Fact ; + mem:content "Merge-preview changes pagination: subjects sorted by full IRI (decode_sid) — that ordering IS the cursor contract (changes_after_subject resumes strictly after). max_changes counts flakes, cuts at subject boundaries, first subject of a page returned whole even over cap; Some(0)=stats-only (exact counts, truncated=true, no source-state load); CLI --stat maps to Some(0) while CLI --max-changes 0 keeps the CLI's 0=unbounded convention (None locally, param omitted remotely). Source-side walk is shared: compute_delta_keys_and_changes yields conflict keys + netted flakes in one trace_commits_by_id pass (iterate commit.flakes.rev() so within-commit order stays reverse-chronological)." ; + mem:tag "branches" ; + mem:tag "cli" ; + mem:tag "merge-preview" ; + mem:tag "pagination" ; + mem:scope mem:repo ; + mem:artifactRef "fluree-db-api/src/merge_preview.rs" ; + mem:artifactRef "fluree-db-cli/src/commands/branch.rs" ; + mem:artifactRef "fluree-db-novelty/src/delta.rs" ; + mem:branch "feature/merge-preview-changes" ; + mem:createdAt "2026-07-05T17:15:17.046896+00:00"^^xsd:dateTime . + mem:constraint-01kwjtjew74a727brs0q3jaf7e a mem:Constraint ; mem:content "SHACL tx-path engine identity: fluree-db-api builds a ShaclEngine but hands only the ShaclCache across the crate boundary into fluree-db-transact's validate_view_with_shacl, which used to rebuild a HIERARCHY-LESS engine — engine-level state (hierarchy for subproperty/predicate-target entailment) silently vanished at transaction time while cache-baked state (subclass target index) survived. Fixed: validate_view_with_shacl takes Arc + Option via ShaclEngine::from_shared_cache. When adding engine-level fields, thread them through this boundary or they only apply to validate_all paths." ; mem:tag "crate-boundary" ; diff --git a/docs/api/endpoints.md b/docs/api/endpoints.md index c135de950..19faaf5ae 100644 --- a/docs/api/endpoints.md +++ b/docs/api/endpoints.md @@ -2335,7 +2335,7 @@ Bearer token required when `data_auth.mode = required`; reads are gated on `bear **URL:** ``` -GET /merge-preview/{ledger-name}?source={source}&target={target}&max_commits={n}&max_conflict_keys={n}&include_conflicts={bool}&include_conflict_details={bool}&strategy={strategy} +GET /merge-preview/{ledger-name}?source={source}&target={target}&max_commits={n}&max_conflict_keys={n}&include_conflicts={bool}&include_conflict_details={bool}&strategy={strategy}&include_changes={bool}&max_changes={n}&changes_after_subject={iri} ``` **Path / Query Parameters:** @@ -2350,6 +2350,9 @@ GET /merge-preview/{ledger-name}?source={source}&target={target}&max_commits={n} | `include_conflicts` | bool | No | When false, skips the conflict computation (default true). Use this to make the preview cheap on diverged branches. | | `include_conflict_details` | bool | No | When true, includes source/target flake values for the returned conflict keys. Defaults to false. Details are computed after `max_conflict_keys` is applied. | | `strategy` | string | No | Strategy used to annotate conflict details. Defaults to `take-both`. Options: `take-both`, `abort`, `take-source`, `take-branch`. | +| `include_changes` | bool | No | When true, includes the aggregate **netted** change set the merge would apply (source side, ancestor..source-head) as `changes`. Defaults to false. Costs one full commit load per commit in the source divergence; the walk is shared with the conflict computation when both are requested. | +| `max_changes` | number | No | Cap on change entries returned, counted in **flakes** and cut at subject boundaries (default 500; server clamps to a hard maximum of 5,000). A single subject larger than the cap is returned whole. `0` is a valid "diff stats" mode: exact counts, no payload. Bounds response size, **not** the replay walk. | +| `changes_after_subject` | string | No | Pagination cursor: return only subjects whose full IRI sorts strictly after this value. Pass the previous response's `changes.next_cursor`. Each page re-pays the full replay + netting cost. Requires `include_changes=true`. | **Response body (200 OK):** @@ -2386,6 +2389,19 @@ GET /merge-preview/{ledger-name}?source={source}&target={target}&max_commits={n} } } ] + }, + "changes": { + "assert_count": 2, + "retract_count": 1, + "subject_count": 1, + "entries": [ + { + "subject": "http://example.org/ns/alice", + "asserts": [["ex:alice", "ex:status", "active", "xsd:string", true]], + "retracts": [["ex:alice", "ex:status", "archived", "xsd:string", false]] + } + ], + "truncated": false } } ``` @@ -2400,15 +2416,23 @@ GET /merge-preview/{ledger-name}?source={source}&target={target}&max_commits={n} | `fast_forward` | bool | True when target HEAD == ancestor (or both heads absent) | | `mergeable` | bool | False only when the selected preview strategy would abort, e.g. `strategy=abort` with conflicts. This is a strategy/conflict signal, not full transaction validation. `mergeable=true` does not guarantee a subsequent `POST /merge` will succeed; it only reflects the conflict/strategy interaction at preview time. | | `conflicts` | object | Overlapping `(s, p, g)` keys touched on both sides since the ancestor. Empty when `fast_forward` or `include_conflicts=false` | +| `changes` | object | Present iff `include_changes=true`. Aggregate netted change set — see below | Per-commit summaries (`ahead.commits[]` / `behind.commits[]`) are newest-first and include assert/retract counts plus an optional `message` extracted from `txn_meta` when an `f:message` string entry is present. When `include_conflict_details=true`, `conflicts.details[]` contains one entry for each returned conflict key. `source_values` and `target_values` are the current asserted values for that key at each branch HEAD, using the same resolved flake tuple format as `/show`: `[subject, predicate, object, datatype, operation]`, with an optional metadata object as the 6th tuple item. The `resolution` object is an annotation only; preview does not apply the strategy or mutate state. +**The `changes` object** is a git-diff-style rollup of the merge: the source side's `ancestor..source_head` flakes folded per fact (full identity: subject, predicate, object, datatype, graph, language tag, list index), with internally-cancelling assert/retract pairs removed. Intermediate churn — a fact created then deleted within the range, or deleted then restored — never appears. This is the *net commit effect*: what replaying the source range applies, minus pairs that cancel, so a re-assert of a value that already existed before the range still nets as an assert. The change set is strategy-independent (the raw source-vs-ancestor delta, before conflict resolution); under a non-default strategy, conflicting keys resolve per `conflicts.details`. + +- `assert_count` / `retract_count` / `subject_count` are exact across the full divergence, never truncated — a UI can render "showing X of Y". +- `entries[]` groups net changes by subject, subjects ordered by full IRI. Each flake uses the same resolved tuple format as conflict details. +- `truncated` is true when subjects were withheld by `max_changes` (or when `max_changes=0` suppressed the payload). +- `next_cursor` (present only when truncated by the cap) is the last returned subject IRI; pass it as `changes_after_subject` to fetch the next page. + **Status codes:** - `200 OK` — Preview computed successfully -- `400 Bad Request` — Source has no branch point (e.g., main), `source == target`, unknown strategy, unsupported preview strategy, `include_conflict_details=true` with `include_conflicts=false`, or `strategy=abort` with `include_conflicts=false` +- `400 Bad Request` — Source has no branch point (e.g., main), `source == target`, unknown strategy, unsupported preview strategy, `include_conflict_details=true` with `include_conflicts=false`, `strategy=abort` with `include_conflicts=false`, or `changes_after_subject` without `include_changes=true` - `401 Unauthorized` — Bearer token required - `404 Not Found` — Ledger or branch does not exist (or bearer cannot read it) @@ -2426,6 +2450,15 @@ curl "http://localhost:8090/v1/fluree/merge-preview/mydb?source=dev&max_commits= # Include value details and labels for a source-winning merge curl "http://localhost:8090/v1/fluree/merge-preview/mydb?source=dev&target=main&include_conflict_details=true&strategy=take-source" + +# Aggregate net change set for a merge-request "Changes" panel +curl "http://localhost:8090/v1/fluree/merge-preview/mydb?source=dev&include_changes=true" + +# Cheap diff stats: exact net counts, no payload +curl "http://localhost:8090/v1/fluree/merge-preview/mydb?source=dev&include_changes=true&max_changes=0" + +# Next page of a large diff +curl "http://localhost:8090/v1/fluree/merge-preview/mydb?source=dev&include_changes=true&changes_after_subject=http%3A%2F%2Fexample.org%2Fns%2Falice" ``` ### GET /info/{ledger-id} diff --git a/docs/cli/server-integration.md b/docs/cli/server-integration.md index 435d07a7e..1e617f4f8 100644 --- a/docs/cli/server-integration.md +++ b/docs/cli/server-integration.md @@ -378,14 +378,42 @@ Same admin auth bracket as `/create`, `/drop`, `/reindex`. See ### `fluree branch diff` (read-only merge preview) -- `GET {api_base_url}/merge-preview/*ledger?source=&target=&max_commits=&max_conflict_keys=&include_conflicts=` +- `GET {api_base_url}/merge-preview/*ledger?source=&target=&max_commits=&max_conflict_keys=&include_conflicts=&include_changes=&max_changes=&changes_after_subject=` Returns the rich diff between two branches — ahead/behind commit summaries, -common ancestor, conflict keys, fast-forward eligibility — without mutating -any nameservice or content-store state. See +common ancestor, conflict keys, fast-forward eligibility, and (opt-in) the +aggregate netted change set — without mutating any nameservice or +content-store state. See [Merge Preview Contract](#merge-preview-contract) for the full semantic and response-shape spec. +**Using the CLI from external apps.** Applications that shell out to the +CLI (instead of calling the HTTP endpoint directly) get the same diff +through `fluree branch diff`: + +```bash +# Machine-readable: --json emits the raw preview (identical shape to the +# HTTP response body, including the `changes` object when requested) +fluree branch diff dev --changes --json --remote origin -l mydb + +# Cheap stats-only probe (exact net counts, no payload) +fluree branch diff dev --stat --json --remote origin -l mydb + +# Page a large diff: read changes.next_cursor from the previous output +fluree branch diff dev --changes --json --changes-after '' \ + --remote origin -l mydb +``` + +The CLI resolves the same three modes everywhere: `--remote ` targets +a configured remote server, tracked ledgers route through their tracking +remote automatically, and plain local ledgers compute the preview in-process +(no server required). In all modes `--json` output is the `MergePreview` +JSON documented below, so an app can parse one shape regardless of where +the ledger lives. Errors surface as a nonzero exit code with a message on +stderr. Note the CLI cap convention: `--max-changes 0` means *unbounded* +(local mode only; over HTTP the server's cap still applies) — stats-only +mode is spelled `--stat`, which maps to `max_changes=0` on the wire. + ## Policy Enforcement Contract CLI policy flags ride on every data API request as both HTTP headers and (for @@ -647,6 +675,7 @@ terminal `end` record arrive in order. GET {api_base_url}/merge-preview/{ledger}?source={source}&target={target} &max_commits={n}&max_conflict_keys={n}&include_conflicts={bool} &include_conflict_details={bool}&strategy={strategy} + &include_changes={bool}&max_changes={n}&changes_after_subject={iri} ``` | Parameter | Type | Required | Server default | Description | @@ -659,6 +688,9 @@ GET {api_base_url}/merge-preview/{ledger}?source={source}&target={target} | `include_conflicts` | bool | No | `true` | When `false`, the conflict computation is skipped | | `include_conflict_details` | bool | No | `false` | When `true`, include source/target flake values for the returned conflict keys | | `strategy` | string | No | `take-both` | Strategy used for resolution labels in `conflicts.details[].resolution`; one of `take-both`, `abort`, `take-source`, `take-branch` | +| `include_changes` | bool | No | `false` | When `true`, include the aggregate netted change set as `changes` | +| `max_changes` | integer | No | `500` | Cap on `changes.entries`, counted in flakes, cut at subject boundaries. `0` = stats-only mode | +| `changes_after_subject` | string | No | — | Pagination cursor (full subject IRI); requires `include_changes=true` | Auth follows the same pattern as `GET /branch/*ledger` (read-only): require a Bearer when `data_auth.mode == required`; gate on `can_read(ledger)`; @@ -733,6 +765,28 @@ These rules are not negotiable; the CLI and other clients depend on them: example, reject when `target.t - ancestor.t` exceeds some threshold) or document that clients should pass `include_conflicts=false` for a cheaper preview. +11. **Aggregate change set.** When `include_changes == true`, populate + `changes` with the source side's `ancestor..source_head` flakes **netted + per fact** — full fact identity is `(subject, predicate, object, + datatype, graph, language tag, list index)`; a fact survives only when + its oldest and newest in-range ops agree (net op = newest op), so + create-then-delete and delete-then-restore churn never appears. The set + is strategy-independent (raw source-vs-ancestor delta, before conflict + resolution). `assert_count` / `retract_count` / `subject_count` are + exact and unaffected by the cap. `entries` groups changes by subject, + subjects ordered by **full IRI** (this ordering is the pagination + contract); flakes use the same resolved tuple shape as conflict + details. The `max_changes` cap counts flakes but cuts at subject + boundaries — never split a subject across pages; a single subject + larger than the cap is returned whole. `max_changes=0` is stats-only + (exact counts, empty `entries`, `truncated=true` when changes exist). + When truncated by the cap, `next_cursor` is the last returned subject + IRI; `changes_after_subject` resumes strictly after it. The reference + server clamps `max_changes` with hard max `5_000` + (`PREVIEW_HARD_MAX_CHANGES`). `changes_after_subject` without + `include_changes=true` is a `400`. The source-side commit replay is + shared with the conflict walk when both are requested; each pagination + page re-pays the replay cost. ### Response (`200 OK`) @@ -777,6 +831,21 @@ These rules are not negotiable; the CLI and other clients depend on them: } } ] + }, + // present iff include_changes=true + "changes": { + "assert_count": 2, + "retract_count": 1, + "subject_count": 1, + "entries": [ + { + "subject": "http://example.org/ns/alice", + "asserts": [["ex:alice", "ex:status", "active", "xsd:string", true]], + "retracts": [["ex:alice", "ex:status", "archived", "xsd:string", false]] + } + ], + "truncated": false + // "next_cursor": "" — only when truncated by the cap } } ``` diff --git a/docs/getting-started/rust-api.md b/docs/getting-started/rust-api.md index 7e706fa69..44b390d7e 100644 --- a/docs/getting-started/rust-api.md +++ b/docs/getting-started/rust-api.md @@ -1146,6 +1146,7 @@ async fn main() -> Result<()> { max_commits: Some(0), // counts only — no commit summaries max_conflict_keys: Some(0), include_conflicts: false, + ..MergePreviewOpts::default() }, ) .await?; @@ -1162,6 +1163,7 @@ async fn main() -> Result<()> { max_commits: None, max_conflict_keys: None, include_conflicts: true, + ..MergePreviewOpts::default() }, ) .await?; @@ -1170,6 +1172,74 @@ async fn main() -> Result<()> { } ``` +#### The aggregate change set (`include_changes`) + +For a merge-request "Changes" panel, `include_changes` returns the **net** +set of facts the merge would apply — the source side's commits since the +common ancestor folded per fact, with internally-cancelling assert/retract +pairs removed. A branch with 40 commits that ultimately changes 12 facts +reviews as 12 facts: + +```rust +use fluree_db_api::{FlureeBuilder, MergePreviewOpts, Result}; + +#[tokio::main] +async fn main() -> Result<()> { + let fluree = FlureeBuilder::memory().build_memory(); + + // ... create ledger, branch, transact on dev, etc. + + let preview = fluree + .merge_preview_with( + "mydb", + "dev", + None, + MergePreviewOpts { + include_changes: true, + // Cap the payload at 100 flakes, cut at subject boundaries. + // Some(0) = stats-only; None = unbounded (Rust-only escape hatch). + max_changes: Some(100), + ..MergePreviewOpts::default() + }, + ) + .await?; + + let changes = preview.changes.expect("include_changes was set"); + println!( + "+{} -{} across {} subject(s)", + changes.assert_count, changes.retract_count, changes.subject_count, + ); + for entry in &changes.entries { + println!("{}: +{} -{}", entry.subject, entry.asserts.len(), entry.retracts.len()); + } + // Page through a large diff: subjects are ordered by full IRI and a + // truncated page carries a resume cursor. + if let Some(cursor) = changes.next_cursor { + let _next_page = fluree + .merge_preview_with( + "mydb", + "dev", + None, + MergePreviewOpts { + include_changes: true, + max_changes: Some(100), + changes_after_subject: Some(cursor), + ..MergePreviewOpts::default() + }, + ) + .await?; + } + Ok(()) +} +``` + +Exact counts (`assert_count` / `retract_count` / `subject_count`) are never +truncated, so a UI can render "showing X of Y". The change set is +strategy-independent — the raw source-vs-ancestor delta before conflict +resolution; conflicting keys resolve per `conflicts.details`. Each page +re-pays the source-side replay cost (there is no cache), matching what the +merge itself would pay. + #### What the caps do (and don't) control `max_commits` and `max_conflict_keys` cap the **size of the returned diff --git a/fluree-db-api/src/lib.rs b/fluree-db-api/src/lib.rs index 51f66e15d..fa9d55095 100644 --- a/fluree-db-api/src/lib.rs +++ b/fluree-db-api/src/lib.rs @@ -159,8 +159,8 @@ pub use ledger_manager::{ pub use ledger_view::{CommitRef, LedgerView}; pub use merge::{MergeReport, StagedMerge}; pub use merge_preview::{ - AncestorRef, BranchDelta, ConflictDetail, ConflictResolutionPreview, ConflictSummary, - MergePreview, MergePreviewOpts, + AncestorRef, BranchDelta, ChangeSummary, ConflictDetail, ConflictResolutionPreview, + ConflictSummary, MergePreview, MergePreviewOpts, SubjectChange, DEFAULT_MAX_CHANGES, }; pub use pack::{ compute_missing_index_artifacts, full_ledger_pack_request, validate_pack_request, PackChunk, diff --git a/fluree-db-api/src/merge_preview.rs b/fluree-db-api/src/merge_preview.rs index 80039c5cd..5092f20c0 100644 --- a/fluree-db-api/src/merge_preview.rs +++ b/fluree-db-api/src/merge_preview.rs @@ -20,8 +20,9 @@ use fluree_db_core::{ ContentId, ContentStore, Flake, }; use fluree_db_ledger::LedgerState; -use fluree_db_novelty::compute_delta_keys; +use fluree_db_novelty::{compute_delta_keys, compute_delta_keys_and_changes}; use futures::{stream, StreamExt, TryStreamExt}; +use rustc_hash::{FxHashMap, FxHashSet}; use serde::Serialize; use std::sync::Arc; use tracing::Instrument; @@ -32,6 +33,10 @@ pub const DEFAULT_MAX_COMMITS: usize = 500; /// Default cap on conflict keys returned in [`ConflictSummary::keys`]. pub const DEFAULT_MAX_CONFLICT_KEYS: usize = 200; +/// Default cap on change entries returned in [`ChangeSummary::entries`], +/// counted in flakes. +pub const DEFAULT_MAX_CHANGES: usize = 500; + /// Knobs for [`crate::Fluree::merge_preview`]. /// /// `MergePreviewOpts::default()` matches the spec: cap each commit list at @@ -76,6 +81,26 @@ pub struct MergePreviewOpts { pub include_conflict_details: bool, /// Strategy used for human-readable conflict resolution labels. pub conflict_strategy: ConflictStrategy, + /// When `true`, include the aggregate netted change set the merge would + /// apply (source side, `ancestor..source_head`) as + /// [`MergePreview::changes`]. Costs one full commit load per commit in + /// the source divergence; the walk is shared with the conflict + /// computation when both are requested. + pub include_changes: bool, + /// Cap on change entries returned in [`ChangeSummary::entries`], counted + /// in **flakes** (not subjects) and cut at subject boundaries — a + /// subject's diff is never split across the cap, so a single subject + /// larger than the cap is returned whole (bounded overshoot). `Some(0)` + /// is a valid "diff stats" mode: exact counts, no payload, and no + /// source-state load for IRI resolution. `None` is unbounded. Caps the + /// response size only — counts stay exact and the replay walk is + /// unaffected. + pub max_changes: Option, + /// Resume cursor for paging [`ChangeSummary::entries`]: return only + /// subjects whose full IRI sorts strictly after this value. Pass the + /// previous response's [`ChangeSummary::next_cursor`]. Each page re-pays + /// the full replay + netting cost. Requires `include_changes`. + pub changes_after_subject: Option, } impl Default for MergePreviewOpts { @@ -86,6 +111,9 @@ impl Default for MergePreviewOpts { include_conflicts: true, include_conflict_details: false, conflict_strategy: ConflictStrategy::default(), + include_changes: false, + max_changes: Some(DEFAULT_MAX_CHANGES), + changes_after_subject: None, } } } @@ -155,6 +183,70 @@ impl ConflictSummary { } } +/// Aggregate netted change set a merge would apply — the source side's +/// `ancestor..source_head` flakes folded per fact, with +/// internally-cancelling assert/retract pairs removed. +/// +/// ### Netting contract +/// +/// Per fact (full identity: subject, predicate, object, datatype, graph, +/// language tag, list index), the net op is the newest in-range op; a fact +/// survives only when its oldest and newest in-range ops agree. Intermediate +/// churn (create-then-delete, delete-then-restore) never appears. This is +/// the **net commit effect** — what replaying the range applies, minus +/// pairs that cancel — so a re-assert of a value that already existed +/// before the range still nets as an assert (the merge does apply it). +/// +/// The change set is strategy-independent: it is the raw source-vs-ancestor +/// delta, *before* conflict resolution. Under a non-default strategy, +/// conflicting keys resolve per the separately-returned +/// [`ConflictSummary::details`]. +#[derive(Clone, Debug, Serialize)] +pub struct ChangeSummary { + /// Exact net assert count across the full divergence. Never truncated. + pub assert_count: usize, + /// Exact net retract count across the full divergence. Never truncated. + pub retract_count: usize, + /// Exact number of distinct subjects touched by net changes. + pub subject_count: usize, + /// Net changes grouped by subject, subjects ordered by full IRI. + /// Bounded by [`MergePreviewOpts::max_changes`] (counting flakes, cut at + /// subject boundaries) and offset by + /// [`MergePreviewOpts::changes_after_subject`]. + pub entries: Vec, + /// `true` when subjects after this page were withheld — either by the + /// flake cap or because `max_changes = 0` suppressed the payload. + pub truncated: bool, + /// Cursor for the next page when `truncated` by the flake cap: the last + /// returned subject IRI, to be passed as `changes_after_subject`. + #[serde(skip_serializing_if = "Option::is_none")] + pub next_cursor: Option, +} + +impl ChangeSummary { + fn empty() -> Self { + Self { + assert_count: 0, + retract_count: 0, + subject_count: 0, + entries: Vec::new(), + truncated: false, + next_cursor: None, + } + } +} + +/// Net changes for one subject. +#[derive(Clone, Debug, Serialize)] +pub struct SubjectChange { + /// Full (expanded) subject IRI — also the pagination sort key. + pub subject: String, + /// Net additions on this subject. + pub asserts: Vec, + /// Net removals on this subject. + pub retracts: Vec, +} + /// Read-only diff between two branches. #[derive(Clone, Debug, Serialize)] pub struct MergePreview { @@ -179,6 +271,11 @@ pub struct MergePreview { /// Whether the selected strategy can be applied without aborting. pub mergeable: bool, + + /// Aggregate netted change set. Present iff + /// [`MergePreviewOpts::include_changes`] was set. + #[serde(skip_serializing_if = "Option::is_none")] + pub changes: Option, } impl crate::Fluree { @@ -245,6 +342,11 @@ impl crate::Fluree { "Skip strategy is not supported for merge preview".to_string(), )); } + if opts.changes_after_subject.is_some() && !opts.include_changes { + return Err(ApiError::InvalidBranch( + "changes_after_subject requires include_changes=true".to_string(), + )); + } // ---- Resolve records (mirrors merge_branch_inner). ---------------- let source_id = format_ledger_id(ledger_name, source_branch); @@ -366,23 +468,59 @@ impl crate::Fluree { commits: behind_summaries, }; - // ---- Conflicts (only if relevant). -------------------------------- - let conflicts = if !opts.include_conflicts || fast_forward { - ConflictSummary::empty() - } else { - match (&source_head, &target_head, &ancestor) { - (Some(s_head), Some(t_head), Some(anc)) => { - let s_delta_fut = - compute_delta_keys(source_store.clone(), s_head.clone(), anc.t); - let t_delta_fut = - compute_delta_keys(target_branched.clone(), t_head.clone(), anc.t); - let (s_delta, t_delta) = tokio::try_join!(s_delta_fut, t_delta_fut)?; + // ---- Source-side replay (conflicts and/or changes). --------------- + // Conflicts and the netted change set both fold the source commit + // chain since the ancestor; when both are requested one walk serves + // both. Conflicts are only possible on non-fast-forward merges; + // changes are meaningful regardless (fast-forward is the common + // merge-request review case). + let need_conflicts = opts.include_conflicts + && !fast_forward + && source_head.is_some() + && target_head.is_some() + && ancestor.is_some(); + let need_changes = opts.include_changes; + + let source_fut = async { + match &source_head { + Some(s_head) if need_changes => { + let (keys, net) = compute_delta_keys_and_changes( + source_store.clone(), + s_head.clone(), + stop_at_t, + ) + .await?; + Ok::<_, ApiError>((Some(keys), Some(net))) + } + Some(s_head) if need_conflicts => { + let keys = + compute_delta_keys(source_store.clone(), s_head.clone(), stop_at_t).await?; + Ok((Some(keys), None)) + } + _ => Ok((None, None)), + } + }; + let target_fut = async { + match (&target_head, &ancestor) { + (Some(t_head), Some(anc)) if need_conflicts => { + let keys = + compute_delta_keys(target_branched.clone(), t_head.clone(), anc.t).await?; + Ok::<_, ApiError>(Some(keys)) + } + _ => Ok(None), + } + }; + let ((source_delta, net_flakes), target_delta) = tokio::try_join!(source_fut, target_fut)?; + // ---- Conflicts (only if relevant). -------------------------------- + let (conflict_keys, conflict_count, conflicts_truncated) = + match (need_conflicts, &source_delta, &target_delta) { + (true, Some(s_delta), Some(t_delta)) => { // Sort lexicographically by (s, p, g) so capped responses // are stable across builds and across requests — `HashSet` // intersection order is otherwise unspecified. let mut keys: Vec = - s_delta.intersection(&t_delta).cloned().collect(); + s_delta.intersection(t_delta).cloned().collect(); keys.sort(); let count = keys.len(); let truncated = match opts.max_conflict_keys { @@ -392,46 +530,90 @@ impl crate::Fluree { } _ => false, }; - - let details = if opts.include_conflict_details && !keys.is_empty() { - let source_state_fut = self.load_queryable_state_with_store( - source_store.clone(), - source_record.clone(), - ); - let target_state_fut = self.load_queryable_state_with_store( - target_branched.clone(), - target_record.clone(), - ); - let (source_state, target_state) = - tokio::try_join!(source_state_fut, target_state_fut)?; - - build_conflict_details( - &keys, - &source_state, - &target_state, - &opts.conflict_strategy, - ) - .await? - } else { - Vec::new() - }; - - ConflictSummary { - count, - keys, - truncated, - strategy: if count > 0 { - Some(opts.conflict_strategy.as_str().to_string()) - } else { - None - }, - details, - } + (keys, count, truncated) } - _ => ConflictSummary::empty(), + _ => (Vec::new(), 0, false), + }; + + // ---- Load states needed for IRI resolution. ----------------------- + // Conflict details need both sides; the change payload needs only the + // source side (its head state carries every namespace code the + // divergence introduced). Stats-only mode (`max_changes = 0`) skips + // the load entirely. + let want_details = opts.include_conflict_details && !conflict_keys.is_empty(); + let want_change_payload = need_changes + && net_flakes.as_ref().is_some_and(|n| !n.is_empty()) + && opts.max_changes != Some(0); + + let (source_state, target_state) = if want_details { + let source_state_fut = + self.load_queryable_state_with_store(source_store.clone(), source_record.clone()); + let target_state_fut = self + .load_queryable_state_with_store(target_branched.clone(), target_record.clone()); + let (s, t) = tokio::try_join!(source_state_fut, target_state_fut)?; + (Some(s), Some(t)) + } else if want_change_payload { + let s = self + .load_queryable_state_with_store(source_store.clone(), source_record.clone()) + .await?; + (Some(s), None) + } else { + (None, None) + }; + + let conflicts = if !need_conflicts { + ConflictSummary::empty() + } else { + let details = if want_details { + build_conflict_details( + &conflict_keys, + source_state.as_ref().expect("loaded for details"), + target_state.as_ref().expect("loaded for details"), + &opts.conflict_strategy, + ) + .await? + } else { + Vec::new() + }; + + ConflictSummary { + count: conflict_count, + keys: conflict_keys, + truncated: conflicts_truncated, + strategy: if conflict_count > 0 { + Some(opts.conflict_strategy.as_str().to_string()) + } else { + None + }, + details, } }; + // ---- Changes (only if requested). ---------------------------------- + let changes = if need_changes { + // No compactor in stats-only mode even when conflict details + // loaded the source state — `max_changes = 0` means no payload. + let compactor = if want_change_payload { + source_state + .as_ref() + .map(|s| IriCompactor::from_namespaces(s.snapshot.shared_namespaces())) + } else { + None + }; + let summary = match net_flakes { + Some(net) if !net.is_empty() => build_change_summary( + net, + compactor, + opts.max_changes, + opts.changes_after_subject.as_deref(), + )?, + _ => ChangeSummary::empty(), + }; + Some(summary) + } else { + None + }; + let mergeable = opts.conflict_strategy != ConflictStrategy::Abort || conflicts.count == 0; // ---- Invariants (debug-only). ------------------------------------- @@ -457,10 +639,114 @@ impl crate::Fluree { fast_forward, conflicts, mergeable, + changes, }) } } +/// Group netted flakes by subject, order deterministically, and apply the +/// cursor + flake cap. Counts are computed over the full net set before any +/// truncation. `compactor: None` is stats-only mode — exact counts, no +/// entries. +fn build_change_summary( + net_flakes: Vec, + compactor: Option, + max_changes: Option, + after_subject: Option<&str>, +) -> Result { + let assert_count = net_flakes.iter().filter(|f| f.op).count(); + let retract_count = net_flakes.len() - assert_count; + + let Some(compactor) = compactor else { + // Stats-only: distinct subjects countable without IRI resolution + // (Sid ↔ IRI is bijective under one namespace table). + let subjects: FxHashSet<&fluree_db_core::Sid> = net_flakes.iter().map(|f| &f.s).collect(); + return Ok(ChangeSummary { + assert_count, + retract_count, + subject_count: subjects.len(), + entries: Vec::new(), + truncated: !net_flakes.is_empty(), + next_cursor: None, + }); + }; + + // Group by Sid first — a cheap Arc-clone key with no IRI resolution — so + // each distinct subject is decoded exactly once rather than once per flake. + // Sorting by the decoded IRI then gives the deterministic subject order + // that pagination cursors rely on. + let mut by_sid: FxHashMap> = FxHashMap::default(); + for f in net_flakes { + by_sid.entry(f.s.clone()).or_default().push(f); + } + let subject_count = by_sid.len(); + + let mut by_subject: Vec<(String, Vec)> = by_sid + .into_iter() + .map(|(sid, flakes)| Ok((compactor.decode_sid(&sid)?, flakes))) + .collect::>()?; + by_subject.sort_by(|(a, _), (b, _)| a.cmp(b)); + + let mut entries = Vec::new(); + let mut emitted_flakes = 0usize; + let mut truncated = false; + + for (subject, mut flakes) in by_subject + .into_iter() + .skip_while(|(s, _)| after_subject.is_some_and(|a| s.as_str() <= a)) + { + // Cut at subject boundaries: stop before a subject that would cross + // the cap — unless it's the first subject of the page, which is + // returned whole (bounded overshoot) so progress is always possible. + if let Some(cap) = max_changes { + if !entries.is_empty() && emitted_flakes + flakes.len() > cap { + truncated = true; + break; + } + } + emitted_flakes += flakes.len(); + + // Deterministic order within a subject. + flakes.sort_by(|a, b| { + a.p.cmp(&b.p) + .then_with(|| a.o.cmp(&b.o)) + .then_with(|| a.dt.cmp(&b.dt)) + .then_with(|| a.m.cmp(&b.m)) + }); + + let mut asserts = Vec::new(); + let mut retracts = Vec::new(); + for f in &flakes { + let resolved = resolve_flake(&compactor, f)?; + if f.op { + asserts.push(resolved); + } else { + retracts.push(resolved); + } + } + entries.push(SubjectChange { + subject, + asserts, + retracts, + }); + } + + let next_cursor = if truncated { + entries.last().map(|e| e.subject.clone()) + } else { + None + }; + + Ok(ChangeSummary { + assert_count, + retract_count, + subject_count, + entries, + truncated, + next_cursor, + }) +} + async fn build_conflict_details( keys: &[ConflictKey], source_state: &LedgerState, diff --git a/fluree-db-api/tests/it_merge_preview.rs b/fluree-db-api/tests/it_merge_preview.rs index 892beca58..405573833 100644 --- a/fluree-db-api/tests/it_merge_preview.rs +++ b/fluree-db-api/tests/it_merge_preview.rs @@ -1192,3 +1192,420 @@ async fn preview_conflict_keys_are_sorted() { assert!(pair[0] <= pair[1], "conflict keys must be sorted"); } } + +// ============================================================================= +// 9. Aggregate change set (include_changes) +// ============================================================================= + +/// Standard fixture: `main` holds ex:alice, branch `dev` is created from it. +async fn setup_branch(fluree: &fluree_db_api::Fluree) -> fluree_db_api::LedgerState { + let ledger = fluree.create_ledger("mydb").await.unwrap(); + let base = json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [{"@id": "ex:alice", "ex:name": "Alice"}] + }); + let main_ledger = fluree.insert(ledger, &base).await.unwrap().ledger; + fluree + .create_branch("mydb", "dev", None, None) + .await + .unwrap(); + main_ledger +} + +fn changes_opts() -> MergePreviewOpts { + MergePreviewOpts { + include_changes: true, + ..MergePreviewOpts::default() + } +} + +#[tokio::test] +async fn changes_absent_by_default() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + let preview = fluree.merge_preview("mydb", "dev", None).await.unwrap(); + assert!(preview.changes.is_none()); +} + +#[tokio::test] +async fn changes_fast_forward_basic() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .insert( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [{"@id": "ex:bob", "ex:name": "Bob"}] + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with("mydb", "dev", None, changes_opts()) + .await + .unwrap(); + + assert!(preview.fast_forward); + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 1); + assert_eq!(changes.retract_count, 0); + assert_eq!(changes.subject_count, 1); + assert!(!changes.truncated); + assert!(changes.next_cursor.is_none()); + assert_eq!(changes.entries.len(), 1); + let entry = &changes.entries[0]; + assert_eq!(entry.subject, "http://example.org/ns/bob"); + assert_eq!(entry.asserts.len(), 1); + assert!(entry.retracts.is_empty()); + assert!(entry.asserts[0].op); +} + +#[tokio::test] +async fn changes_netting_cancels_churn() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + // Create bob, then delete him again — the net diff must be empty. + let dev = fluree.ledger("mydb:dev").await.unwrap(); + let dev = fluree + .insert( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [{"@id": "ex:bob", "ex:name": "Bob"}] + }), + ) + .await + .unwrap() + .ledger; + fluree + .update( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "delete": {"@id": "ex:bob", "ex:name": "Bob"} + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with("mydb", "dev", None, changes_opts()) + .await + .unwrap(); + + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 0, "churn must cancel"); + assert_eq!(changes.retract_count, 0); + assert_eq!(changes.subject_count, 0); + assert!(changes.entries.is_empty()); + assert!(!changes.truncated); + // The commits themselves are still visible — only the net diff is empty. + assert!(preview.ahead.count >= 2); +} + +#[tokio::test] +async fn changes_update_shows_both_sides() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + // Rename alice on the branch: net = one retract (old) + one assert (new). + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .update( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "delete": {"@id": "ex:alice", "ex:name": "Alice"}, + "insert": {"@id": "ex:alice", "ex:name": "Alicia"} + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with("mydb", "dev", None, changes_opts()) + .await + .unwrap(); + + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 1); + assert_eq!(changes.retract_count, 1); + assert_eq!(changes.subject_count, 1); + assert_eq!(changes.entries.len(), 1); + let entry = &changes.entries[0]; + assert_eq!(entry.subject, "http://example.org/ns/alice"); + assert_eq!(entry.asserts.len(), 1); + assert_eq!(entry.retracts.len(), 1); +} + +#[tokio::test] +async fn changes_pure_retract_of_ancestor_fact() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .update( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "delete": {"@id": "ex:alice", "ex:name": "Alice"} + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with("mydb", "dev", None, changes_opts()) + .await + .unwrap(); + + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 0); + assert_eq!(changes.retract_count, 1); + assert_eq!(changes.subject_count, 1); + assert!(!changes.entries[0].retracts[0].op); +} + +#[tokio::test] +async fn changes_truncation_cuts_at_subject_boundary_and_pages() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + // Three subjects, two facts each: 6 net flakes total. + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .insert( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [ + {"@id": "ex:s1", "ex:name": "One", "ex:rank": 1}, + {"@id": "ex:s2", "ex:name": "Two", "ex:rank": 2}, + {"@id": "ex:s3", "ex:name": "Three", "ex:rank": 3}, + ] + }), + ) + .await + .unwrap(); + + let page = |after: Option| { + let fluree = &fluree; + async move { + fluree + .merge_preview_with( + "mydb", + "dev", + None, + MergePreviewOpts { + include_changes: true, + max_changes: Some(2), + changes_after_subject: after, + ..MergePreviewOpts::default() + }, + ) + .await + .unwrap() + .changes + .expect("include_changes was set") + } + }; + + // Page 1: cap of 2 flakes fits exactly one subject. + let p1 = page(None).await; + assert_eq!(p1.assert_count, 6, "counts stay exact under the cap"); + assert_eq!(p1.subject_count, 3); + assert_eq!(p1.entries.len(), 1); + assert!(p1.truncated); + let cursor1 = p1.next_cursor.clone().expect("truncated page has cursor"); + assert_eq!(cursor1, p1.entries[0].subject); + + // Page 2 resumes strictly after the cursor. + let p2 = page(Some(cursor1.clone())).await; + assert_eq!(p2.entries.len(), 1); + assert!(p2.entries[0].subject > cursor1, "no overlap between pages"); + assert!(p2.truncated); + + // Page 3 is the last. + let p3 = page(p2.next_cursor.clone()).await; + assert_eq!(p3.entries.len(), 1); + assert!(!p3.truncated); + assert!(p3.next_cursor.is_none()); + + let mut subjects: Vec = [&p1, &p2, &p3] + .iter() + .flat_map(|p| p.entries.iter().map(|e| e.subject.clone())) + .collect(); + let ordered = subjects.clone(); + subjects.sort(); + subjects.dedup(); + assert_eq!(subjects.len(), 3, "pages cover all subjects exactly once"); + assert_eq!(ordered, subjects, "subjects arrive in IRI order"); +} + +#[tokio::test] +async fn changes_stats_only_mode() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .insert( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [ + {"@id": "ex:bob", "ex:name": "Bob"}, + {"@id": "ex:carol", "ex:name": "Carol"}, + ] + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with( + "mydb", + "dev", + None, + MergePreviewOpts { + include_changes: true, + max_changes: Some(0), + ..MergePreviewOpts::default() + }, + ) + .await + .unwrap(); + + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 2); + assert_eq!(changes.retract_count, 0); + assert_eq!(changes.subject_count, 2); + assert!(changes.entries.is_empty()); + assert!(changes.truncated, "payload was withheld"); + assert!(changes.next_cursor.is_none()); +} + +#[tokio::test] +async fn changes_alongside_conflicts_and_details() { + let fluree = FlureeBuilder::memory().build_memory(); + let main_ledger = setup_branch(&fluree).await; + + // Diverge: both sides rename alice. + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .insert( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [{"@id": "ex:alice", "ex:name": "A-dev"}] + }), + ) + .await + .unwrap(); + fluree + .insert( + main_ledger, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [{"@id": "ex:alice", "ex:name": "A-main"}] + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with( + "mydb", + "dev", + None, + MergePreviewOpts { + include_changes: true, + include_conflict_details: true, + ..MergePreviewOpts::default() + }, + ) + .await + .unwrap(); + + assert!(!preview.fast_forward); + assert!(preview.conflicts.count >= 1); + assert!(!preview.conflicts.details.is_empty()); + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 1, "source-side net assert"); + assert_eq!(changes.entries[0].subject, "http://example.org/ns/alice"); +} + +#[tokio::test] +async fn changes_cursor_requires_include_changes() { + let fluree = FlureeBuilder::memory().build_memory(); + setup_branch(&fluree).await; + + let err = fluree + .merge_preview_with( + "mydb", + "dev", + None, + MergePreviewOpts { + changes_after_subject: Some("http://example.org/ns/a".to_string()), + ..MergePreviewOpts::default() + }, + ) + .await + .unwrap_err(); + assert!(err + .to_string() + .contains("changes_after_subject requires include_changes")); +} + +#[tokio::test] +async fn changes_multi_value_predicate_nets_independently() { + let fluree = FlureeBuilder::memory().build_memory(); + let ledger = fluree.create_ledger("mydb").await.unwrap(); + // Alice starts with two nicknames. + fluree + .insert( + ledger, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "@graph": [{"@id": "ex:alice", "ex:nick": ["Al", "Ali"]}] + }), + ) + .await + .unwrap(); + fluree + .create_branch("mydb", "dev", None, None) + .await + .unwrap(); + + // Remove one of the two values on the branch. + let dev = fluree.ledger("mydb:dev").await.unwrap(); + fluree + .update( + dev, + &json!({ + "@context": {"ex": "http://example.org/ns/"}, + "delete": {"@id": "ex:alice", "ex:nick": "Al"} + }), + ) + .await + .unwrap(); + + let preview = fluree + .merge_preview_with("mydb", "dev", None, changes_opts()) + .await + .unwrap(); + + let changes = preview.changes.expect("include_changes was set"); + assert_eq!(changes.assert_count, 0); + assert_eq!( + changes.retract_count, 1, + "only the deleted value appears; the sibling value is untouched" + ); +} diff --git a/fluree-db-cli/src/cli.rs b/fluree-db-cli/src/cli.rs index 2e4702f14..de6505527 100644 --- a/fluree-db-cli/src/cli.rs +++ b/fluree-db-cli/src/cli.rs @@ -1262,6 +1262,26 @@ pub enum BranchAction { #[arg(long)] strategy: Option, + /// Include the aggregate netted change set the merge would apply + /// (git-diff-style rollup, grouped by subject) + #[arg(long)] + changes: bool, + + /// Cap on change entries returned, counted in flakes and cut at + /// subject boundaries (default: 500). Pass 0 for unbounded + /// (local mode only). Implies --changes. + #[arg(long)] + max_changes: Option, + + /// Show only net change counts, no per-fact payload. Implies --changes. + #[arg(long)] + stat: bool, + + /// Pagination cursor: only subjects sorting strictly after this full + /// IRI (from a previous run's next_cursor). Implies --changes. + #[arg(long)] + changes_after: Option, + /// Emit the raw JSON preview instead of a human-readable summary #[arg(long)] json: bool, diff --git a/fluree-db-cli/src/commands/branch.rs b/fluree-db-cli/src/commands/branch.rs index aabada79d..a91cf9adb 100644 --- a/fluree-db-cli/src/commands/branch.rs +++ b/fluree-db-cli/src/commands/branch.rs @@ -103,6 +103,10 @@ pub async fn run(action: BranchAction, dirs: &FlureeDir, direct: bool) -> CliRes no_conflicts, conflict_details, strategy, + changes, + max_changes, + stat, + changes_after, json, ledger, remote, @@ -116,6 +120,13 @@ pub async fn run(action: BranchAction, dirs: &FlureeDir, direct: bool) -> CliRes include_conflicts: !no_conflicts, include_conflict_details: conflict_details, strategy, + include_changes: changes + || stat + || max_changes.is_some() + || changes_after.is_some(), + max_changes, + stat, + changes_after, json, }, ledger.as_deref(), @@ -1031,6 +1042,13 @@ struct DiffOpts { include_conflicts: bool, include_conflict_details: bool, strategy: Option, + include_changes: bool, + /// `None` = default (500). `Some(0)` = unbounded (local mode only), + /// matching the other CLI cap conventions. `--stat` supplies the + /// stats-only mode instead. + max_changes: Option, + stat: bool, + changes_after: Option, json: bool, } @@ -1076,6 +1094,20 @@ async fn run_diff( .as_deref() .or_else(|| include_conflict_details.then_some(conflict_strategy.as_str())); + // `--stat` = stats-only (API `max_changes = 0`); the CLI's `0 = unbounded` + // convention maps to `None` locally and to "omit the param" remotely + // (the server always enforces its own cap). + let include_changes = opts.include_changes; + let max_changes = if opts.stat { + Some(0) + } else { + match opts.max_changes { + None => Some(fluree_db_api::DEFAULT_MAX_CHANGES), + Some(0) => None, + Some(n) => Some(n), + } + }; + if let Some(remote_name) = remote_flag { let alias = context::resolve_ledger(ledger, dirs)?; let (ledger_name, _) = split_ledger_id(&alias)?; @@ -1090,6 +1122,9 @@ async fn run_diff( Some(include_conflicts), Some(include_conflict_details), remote_strategy, + include_changes.then_some(true), + max_changes.filter(|_| include_changes), + opts.changes_after.as_deref(), ) .await?; @@ -1130,6 +1165,9 @@ async fn run_diff( Some(include_conflicts), Some(include_conflict_details), remote_strategy, + include_changes.then_some(true), + max_changes.filter(|_| include_changes), + opts.changes_after.as_deref(), ) .await?; @@ -1149,6 +1187,9 @@ async fn run_diff( include_conflicts, include_conflict_details, conflict_strategy, + include_changes, + max_changes, + changes_after_subject: opts.changes_after.clone(), }; let preview = fluree @@ -1237,6 +1278,41 @@ fn print_preview_local(p: &fluree_db_api::MergePreview) { ); } } + + if let Some(ch) = &p.changes { + let shown: usize = ch + .entries + .iter() + .map(|e| e.asserts.len() + e.retracts.len()) + .sum(); + println!( + "changes: +{} -{} across {} subject(s){}", + ch.assert_count, + ch.retract_count, + ch.subject_count, + if ch.truncated { + format!( + " (showing {} of {} facts)", + shown, + ch.assert_count + ch.retract_count + ) + } else { + String::new() + } + ); + for e in &ch.entries { + println!(" {}", e.subject); + for a in &e.asserts { + println!(" + {}", serde_json::to_string(a).unwrap_or_default()); + } + for r in &e.retracts { + println!(" - {}", serde_json::to_string(r).unwrap_or_default()); + } + } + if let Some(cursor) = &ch.next_cursor { + println!(" next page: --changes-after '{cursor}'"); + } + } } fn print_delta_local(label: &str, d: &fluree_db_api::BranchDelta) { @@ -1352,6 +1428,61 @@ fn print_preview_json(v: &serde_json::Value) -> CliResult<()> { } } } + + if let Some(ch) = v.get("changes").filter(|x| !x.is_null()) { + let asserts = ch.get("assert_count").and_then(Value::as_u64).unwrap_or(0); + let retracts = ch.get("retract_count").and_then(Value::as_u64).unwrap_or(0); + let subjects = ch.get("subject_count").and_then(Value::as_u64).unwrap_or(0); + let truncated = ch + .get("truncated") + .and_then(Value::as_bool) + .unwrap_or(false); + let entries = ch.get("entries").and_then(Value::as_array); + let shown: u64 = entries + .map(|es| { + es.iter() + .map(|e| { + let a = e + .get("asserts") + .and_then(Value::as_array) + .map_or(0, Vec::len); + let r = e + .get("retracts") + .and_then(Value::as_array) + .map_or(0, Vec::len); + (a + r) as u64 + }) + .sum() + }) + .unwrap_or(0); + println!( + "changes: +{asserts} -{retracts} across {subjects} subject(s){}", + if truncated { + format!(" (showing {} of {} facts)", shown, asserts + retracts) + } else { + String::new() + } + ); + if let Some(entries) = entries { + for e in entries { + let subject = e.get("subject").and_then(Value::as_str).unwrap_or("?"); + println!(" {subject}"); + if let Some(list) = e.get("asserts").and_then(Value::as_array) { + for a in list { + println!(" + {}", serde_json::to_string(a).unwrap_or_default()); + } + } + if let Some(list) = e.get("retracts").and_then(Value::as_array) { + for r in list { + println!(" - {}", serde_json::to_string(r).unwrap_or_default()); + } + } + } + } + if let Some(cursor) = ch.get("next_cursor").and_then(Value::as_str) { + println!(" next page: --changes-after '{cursor}'"); + } + } Ok(()) } diff --git a/fluree-db-cli/src/remote_client.rs b/fluree-db-cli/src/remote_client.rs index 8d787d1cb..0a499e774 100644 --- a/fluree-db-cli/src/remote_client.rs +++ b/fluree-db-cli/src/remote_client.rs @@ -2132,7 +2132,7 @@ impl RemoteLedgerClient { /// Read-only merge preview between two branches on the remote server. /// - /// Calls `GET {base_url}/merge-preview/{ledger}?source=&target=&max_commits=&max_conflict_keys=&include_conflicts=&include_conflict_details=&strategy=`. + /// Calls `GET {base_url}/merge-preview/{ledger}?source=&target=&max_commits=&max_conflict_keys=&include_conflicts=&include_conflict_details=&strategy=&include_changes=&max_changes=&changes_after_subject=`. /// The ledger path segment is URL-encoded (via [`op_url`](Self::op_url)) /// so names containing spaces, `?`, `#`, `%`, etc. produce well-formed URLs. #[allow(clippy::too_many_arguments)] @@ -2146,6 +2146,9 @@ impl RemoteLedgerClient { include_conflicts: Option, include_conflict_details: Option, strategy: Option<&str>, + include_changes: Option, + max_changes: Option, + changes_after_subject: Option<&str>, ) -> Result { let mut url = self.op_url("merge-preview", ledger); let mut sep = '?'; @@ -2195,6 +2198,20 @@ impl RemoteLedgerClient { urlencoding::encode(s).into_owned(), ); } + if let Some(b) = include_changes { + push(&mut url, &mut sep, "include_changes", b.to_string()); + } + if let Some(n) = max_changes { + push(&mut url, &mut sep, "max_changes", n.to_string()); + } + if let Some(c) = changes_after_subject { + push( + &mut url, + &mut sep, + "changes_after_subject", + urlencoding::encode(c).into_owned(), + ); + } self.send_json(reqwest::Method::GET, &url, "application/json", None) .await diff --git a/fluree-db-novelty/src/delta.rs b/fluree-db-novelty/src/delta.rs index 9759f306f..1488a94d9 100644 --- a/fluree-db-novelty/src/delta.rs +++ b/fluree-db-novelty/src/delta.rs @@ -5,9 +5,10 @@ //! can be checked against branch commits to detect overlapping changes. use crate::{trace_commits_by_id, Result}; -use fluree_db_core::{ConflictKey, ContentId, ContentStore}; +use fluree_db_core::{ConflictKey, ContentId, ContentStore, Flake, FlakeValue, Sid}; use futures::TryStreamExt; -use rustc_hash::FxHashSet; +use rustc_hash::{FxHashMap, FxHashSet}; +use std::collections::hash_map::Entry; /// Walk the commit chain from `head_id` back to `stop_at_t` and collect /// all (subject, predicate, graph) tuples modified in those commits. @@ -43,3 +44,244 @@ pub async fn compute_delta_keys( Ok(keys) } + +/// Full identity of a fact — every [`Flake`] component except `t` and `op`. +/// +/// This is deliberately NOT `Flake`'s own `Eq`/`Hash` (which ignore `g` by +/// design — flake identity is graph-independent for index purposes). Netting +/// must treat the same triple in two graphs as two distinct facts, and must +/// keep language-tagged strings and list positions apart, so `g`, `lang`, +/// and `i` are all part of the key. +#[derive(PartialEq, Eq, Hash)] +struct FactKey { + g: Option, + s: Sid, + p: Sid, + o: FlakeValue, + dt: Sid, + lang: Option, + i: Option, +} + +impl FactKey { + fn of(flake: &Flake) -> Self { + Self { + g: flake.g.clone(), + s: flake.s.clone(), + p: flake.p.clone(), + o: flake.o.clone(), + dt: flake.dt.clone(), + lang: flake.m.as_ref().and_then(|m| m.lang.clone()), + i: flake.m.as_ref().and_then(|m| m.i), + } + } +} + +struct NetState { + /// Flake of the newest in-range occurrence — the representative emitted + /// when the fact survives netting. + newest: Flake, + /// Op of the oldest in-range occurrence seen so far. + oldest_op: bool, +} + +/// Accumulates the net effect of a commit range on each distinct fact. +/// +/// Feed flakes in strictly **newest-first** order (reverse-chronological). +/// Per fact, the net op is the newest op — but a fact only survives netting +/// when its oldest and newest in-range ops agree: +/// +/// - assert … assert → net assert (created, or re-asserted/replaced) +/// - retract … retract → net retract (removes pre-range state) +/// - assert … retract / retract … assert → dropped (the range ends where +/// it started: created-then-destroyed, or removed-then-restored) +/// +/// This is the "net commit effect" contract: what a merge replaying the +/// range would apply, minus internally-cancelling pairs. It infers pre-range +/// presence from the oldest op, so a re-assert of a value that already +/// existed before the range nets as an assert (the merge does apply it). +#[derive(Default)] +pub struct NetChangeAccumulator { + map: FxHashMap, +} + +impl NetChangeAccumulator { + /// Record one flake. Flakes must arrive newest-first. + pub fn push_newest_first(&mut self, flake: &Flake) { + match self.map.entry(FactKey::of(flake)) { + Entry::Vacant(v) => { + v.insert(NetState { + newest: flake.clone(), + oldest_op: flake.op, + }); + } + // An older occurrence of the same fact: it becomes the oldest. + Entry::Occupied(mut o) => o.get_mut().oldest_op = flake.op, + } + } + + /// Facts that survive netting, each represented by its newest in-range + /// flake (so `op` is the net op). Unordered. + pub fn finish(self) -> Vec { + self.map + .into_values() + .filter(|st| st.oldest_op == st.newest.op) + .map(|st| st.newest) + .collect() + } +} + +/// Like [`compute_delta_keys`], but additionally nets the range's flakes +/// into the aggregate change set the range applies (see +/// [`NetChangeAccumulator`] for the netting contract). +/// +/// One walk serves both outputs, so callers that need conflict keys *and* +/// the change set (merge preview with `include_changes`) replay the source +/// chain once instead of twice. +pub async fn compute_delta_keys_and_changes( + store: C, + head_id: ContentId, + stop_at_t: i64, +) -> Result<(FxHashSet, Vec)> { + let stream = trace_commits_by_id(store, head_id, stop_at_t); + futures::pin_mut!(stream); + + let mut keys = FxHashSet::default(); + let mut acc = NetChangeAccumulator::default(); + + while let Some(commit) = stream.try_next().await? { + // Commits stream newest-first; iterate each commit's flakes in + // reverse so the accumulator sees a strictly reverse-chronological + // sequence even when one commit touches the same fact twice. + for flake in commit.flakes.iter().rev() { + keys.insert(ConflictKey::new( + flake.s.clone(), + flake.p.clone(), + flake.g.clone(), + )); + acc.push_newest_first(flake); + } + } + + Ok((keys, acc.finish())) +} + +#[cfg(test)] +mod tests { + use super::*; + use fluree_db_core::FlakeMeta; + + fn sid(name: &str) -> Sid { + Sid::new(100, name) + } + + fn flake(s: &str, o: &str, t: i64, op: bool) -> Flake { + Flake::new( + sid(s), + sid("pred"), + FlakeValue::String(o.to_string()), + sid("string"), + t, + op, + None, + ) + } + + /// Feed chronological flakes to the accumulator (which expects + /// newest-first) and return the netted result. + fn net(chronological: &[Flake]) -> Vec { + let mut acc = NetChangeAccumulator::default(); + for f in chronological.iter().rev() { + acc.push_newest_first(f); + } + acc.finish() + } + + #[test] + fn lone_assert_survives() { + let out = net(&[flake("a", "v", 1, true)]); + assert_eq!(out.len(), 1); + assert!(out[0].op); + } + + #[test] + fn lone_retract_survives() { + let out = net(&[flake("a", "v", 1, false)]); + assert_eq!(out.len(), 1); + assert!(!out[0].op); + } + + #[test] + fn assert_then_retract_cancels() { + let out = net(&[flake("a", "v", 1, true), flake("a", "v", 2, false)]); + assert!(out.is_empty(), "created-then-destroyed must net to nothing"); + } + + #[test] + fn retract_then_assert_cancels() { + let out = net(&[flake("a", "v", 1, false), flake("a", "v", 2, true)]); + assert!(out.is_empty(), "removed-then-restored must net to nothing"); + } + + #[test] + fn assert_retract_assert_survives_as_assert() { + let out = net(&[ + flake("a", "v", 1, true), + flake("a", "v", 2, false), + flake("a", "v", 3, true), + ]); + assert_eq!(out.len(), 1); + assert!(out[0].op); + assert_eq!(out[0].t, 3, "representative flake is the newest"); + } + + #[test] + fn update_keeps_both_sides() { + // Retract old value + assert new value = two distinct facts, both net. + let out = net(&[flake("a", "old", 2, false), flake("a", "new", 2, true)]); + assert_eq!(out.len(), 2); + } + + #[test] + fn same_commit_pair_cancels_via_reverse_iteration() { + // Both ops at the same t, chronological order assert-then-retract. + let out = net(&[flake("a", "v", 2, true), flake("a", "v", 2, false)]); + assert!(out.is_empty()); + } + + #[test] + fn graph_scoping_keeps_facts_distinct() { + let mut in_graph = flake("a", "v", 1, true); + in_graph.g = Some(sid("g1")); + let out = net(&[flake("a", "v", 1, true), in_graph]); + assert_eq!( + out.len(), + 2, + "same triple in default + named graph must not collapse" + ); + } + + #[test] + fn lang_tags_keep_facts_distinct() { + let mut en = flake("a", "chat", 1, true); + en.m = Some(FlakeMeta::with_lang("en")); + let mut fr = flake("a", "chat", 2, false); + fr.m = Some(FlakeMeta::with_lang("fr")); + let out = net(&[en, fr]); + assert_eq!( + out.len(), + 2, + "differing language tags must not cancel each other" + ); + } + + #[test] + fn list_positions_keep_facts_distinct() { + let mut p0 = flake("a", "v", 1, true); + p0.m = Some(FlakeMeta::with_index(0)); + let mut p1 = flake("a", "v", 1, true); + p1.m = Some(FlakeMeta::with_index(1)); + let out = net(&[p0, p1]); + assert_eq!(out.len(), 2); + } +} diff --git a/fluree-db-novelty/src/lib.rs b/fluree-db-novelty/src/lib.rs index d96a67cb4..ce652a9fe 100644 --- a/fluree-db-novelty/src/lib.rs +++ b/fluree-db-novelty/src/lib.rs @@ -52,7 +52,7 @@ pub use commit::{ pub use commit_flakes::{ generate_commit_flakes, iso_to_epoch_ms_opt, stamp_graph_on_commit_flakes, }; -pub use delta::compute_delta_keys; +pub use delta::{compute_delta_keys, compute_delta_keys_and_changes, NetChangeAccumulator}; pub use error::{NoveltyError, Result}; pub use fluree_db_core::commit::codec::envelope::{MAX_GRAPH_DELTA_ENTRIES, MAX_GRAPH_IRI_LENGTH}; pub use fluree_db_core::commit::codec::format::{CommitSignature, ALGO_ED25519}; diff --git a/fluree-db-server/src/routes/ledger.rs b/fluree-db-server/src/routes/ledger.rs index 1fd02deb8..ce2890c4f 100644 --- a/fluree-db-server/src/routes/ledger.rs +++ b/fluree-db-server/src/routes/ledger.rs @@ -1822,6 +1822,18 @@ const PREVIEW_HARD_MAX_COMMITS: usize = 5_000; /// should pass `include_conflicts=false`. const PREVIEW_HARD_MAX_CONFLICT_KEYS: usize = 5_000; +/// Hard cap on `max_changes`. 10x the recommended default. +/// +/// **What this protects:** the size of `changes.entries` in the response +/// (counted in flakes; cut at subject boundaries, so a single huge subject +/// may overshoot by its own size). +/// +/// **What this does NOT protect:** the change computation. When +/// `include_changes=true`, the source-side commit chain since the ancestor +/// is fully replayed (one commit blob load per commit) regardless of cap — +/// the same cost the merge itself pays. Each pagination page re-pays it. +const PREVIEW_HARD_MAX_CHANGES: usize = 5_000; + /// Query parameters for [`merge_preview`]. #[derive(Deserialize)] pub struct MergePreviewQuery { @@ -1845,6 +1857,17 @@ pub struct MergePreviewQuery { /// Strategy used for conflict resolution labels. Defaults to take-both. #[serde(default)] pub strategy: Option, + /// Include the aggregate netted change set the merge would apply. + #[serde(default)] + pub include_changes: Option, + /// Cap on change entries returned, counted in flakes and cut at subject + /// boundaries. `0` = stats-only (exact counts, no payload). Defaults to 500. + #[serde(default)] + pub max_changes: Option, + /// Pagination cursor: return only subjects sorting strictly after this + /// full IRI. Pass the previous response's `changes.next_cursor`. + #[serde(default)] + pub changes_after_subject: Option, } /// Read-only branch merge preview. @@ -1933,6 +1956,18 @@ pub async fn merge_preview( "strategy=abort requires include_conflicts=true for mergeable preview", )); } + if let Some(b) = params.include_changes { + opts.include_changes = b; + } + if let Some(n) = params.max_changes { + opts.max_changes = Some(n.min(PREVIEW_HARD_MAX_CHANGES)); + } + if params.changes_after_subject.is_some() && !opts.include_changes { + return Err(ServerError::bad_request( + "changes_after_subject requires include_changes=true", + )); + } + opts.changes_after_subject = params.changes_after_subject.clone(); let preview = state .fluree diff --git a/fluree-db-server/tests/integration.rs b/fluree-db-server/tests/integration.rs index 8aa2ddd3a..9be9aafdf 100644 --- a/fluree-db-server/tests/integration.rs +++ b/fluree-db-server/tests/integration.rs @@ -475,6 +475,187 @@ async fn create_branch_at_historical_t() { assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } +#[tokio::test] +async fn merge_preview_include_changes() { + let (_tmp, state) = test_state().await; + let app = build_router(state.clone()); + + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/fluree/create") + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({"ledger": "chg:main"}).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/fluree/insert") + .header("content-type", "application/json") + .header("fluree-ledger", "chg:main") + .body(Body::from( + serde_json::json!({ + "@context": {"ex": "http://example.org/"}, + "@id": "ex:alice", + "ex:name": "Alice", + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/fluree/branch") + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({"ledger": "chg", "branch": "dev"}).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/fluree/insert") + .header("content-type", "application/json") + .header("fluree-ledger", "chg:dev") + .body(Body::from( + serde_json::json!({ + "@context": {"ex": "http://example.org/"}, + "@id": "ex:bob", + "ex:name": "Bob", + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + // Without include_changes the field is absent. + let resp = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/fluree/merge-preview/chg?source=dev") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let (status, json) = json_body(resp).await; + assert_eq!(status, StatusCode::OK); + assert!(json.get("changes").is_none(), "unexpected changes: {json}"); + + // With include_changes the netted change set is present. + let resp = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/fluree/merge-preview/chg?source=dev&include_changes=true") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let (status, json) = json_body(resp).await; + assert_eq!(status, StatusCode::OK); + let changes = json.get("changes").expect("changes present"); + assert_eq!( + changes + .get("assert_count") + .and_then(serde_json::Value::as_u64), + Some(1), + "unexpected changes: {changes}" + ); + assert_eq!( + changes + .get("subject_count") + .and_then(serde_json::Value::as_u64), + Some(1) + ); + let entries = changes.get("entries").and_then(|v| v.as_array()).unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!( + entries[0].get("subject").and_then(|v| v.as_str()), + Some("http://example.org/bob") + ); + + // Stats-only mode: exact counts, no payload. + let resp = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/fluree/merge-preview/chg?source=dev&include_changes=true&max_changes=0") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let (status, json) = json_body(resp).await; + assert_eq!(status, StatusCode::OK); + let changes = json.get("changes").expect("changes present"); + assert_eq!( + changes + .get("assert_count") + .and_then(serde_json::Value::as_u64), + Some(1) + ); + assert_eq!( + changes + .get("entries") + .and_then(|v| v.as_array()) + .map(Vec::len), + Some(0) + ); + assert_eq!( + changes + .get("truncated") + .and_then(serde_json::Value::as_bool), + Some(true) + ); + + // Cursor without include_changes is a 400. + let resp = app + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/fluree/merge-preview/chg?source=dev&changes_after_subject=x") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + #[tokio::test] async fn ledger_with_slash_works_via_op_prefixed_routes() { let (_tmp, state) = test_state().await;