diff --git a/README.md b/README.md index eed9d11de..95d8fe5a5 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,9 @@ Write the domain once, compose it into one `Service` or several, then generate the client. Each stage below uses real code from [`tests/e2e-ui`](tests/e2e-ui). +Unsigned command fields preserve their Rust range in generated client validation +and optimistic projections; see [unsigned command inputs](docs/unsigned-command-inputs.md). + ```mermaid sequenceDiagram actor Author @@ -1054,13 +1057,27 @@ fn upcast_initialized_v1_v2((id, task): InitV1) -> InitV2 { } aggregate!(Todo, entity { - "initialized"(id, task, priority) => initialize, + "initialized"(id, task, priority), version = 2 => initialize, "completed"() => complete(), } upcasters [ ("initialized", 1 => 2, InitV1 => InitV2, upcast_initialized_v1_v2), ]); ``` +Replay requires each event's schema version to match its registered handler +after upcasting. `#[sourced]` reads this version from `#[event(..., version = N)]`; +`aggregate!` declares it on the registration as above and must match the +corresponding `#[digest(..., version = N)]`. Omitted versions default to 1. +Older events need an explicit upcaster chain to the registered version; future +versions are rejected. Matching payload layouts do not bypass this check. +The check runs before payload decoding or handler invocation, including when +loading only the event tail after a snapshot in a native repository or cell. + +**Breaking change:** an `aggregate!` registration for a versioned handler must +now declare its current version. Histories with unsupported versions fail replay +instead of being interpreted using the current payload shape. Stored events are +unchanged; add the appropriate upcasters to read supported older versions. + ## Event Metadata Metadata lets you attach cross-cutting context — correlation IDs, causation IDs, user context, trace spans — to events without changing your domain model. diff --git a/distributed_cli/src/client_compiler/command_manifest/shape.rs b/distributed_cli/src/client_compiler/command_manifest/shape.rs index 09eba1a78..ef20cb8e2 100644 --- a/distributed_cli/src/client_compiler/command_manifest/shape.rs +++ b/distributed_cli/src/client_compiler/command_manifest/shape.rs @@ -219,6 +219,12 @@ fn validate_codec( ) -> Result<(), ClientCompileError> { match scalar_codecs.get(scalar) { Some(expected) if expected == codec => Ok(()), + Some(_) + if scalar == "BigInt" + && matches!(codec, "uint8" | "uint16" | "uint32" | "uint64_safe_integer") => + { + Ok(()) + } Some(expected) => Err(invalid( "client.manifest.command_type_codec", format!("{label} codec `{codec}` does not match `{scalar}` codec `{expected}`"), diff --git a/distributed_cli/src/client_compiler/manifest/mod.rs b/distributed_cli/src/client_compiler/manifest/mod.rs index 92e747912..5eac2d648 100644 --- a/distributed_cli/src/client_compiler/manifest/mod.rs +++ b/distributed_cli/src/client_compiler/manifest/mod.rs @@ -18,7 +18,7 @@ pub(crate) use util::{validate_hash, validate_nonempty}; const MANIFEST_VERSION: u64 = 2; const PROTOCOL_VERSION: u64 = 1; const PROTOCOL_FINGERPRINT: &str = - "sha256:00fb342f3acb4dc1c1716a43cc3001c748d5f6c500ff831690d820e9e43e2782"; + "sha256:0dfa8a3f49e17d8d99c5c095c1ed14f528cae3e55fbc2f2b852975a50936ec5b"; pub(crate) const CLIENT_PROJECTION_PROGRAM_VERSION: u32 = 2; pub(crate) const CLIENT_PROJECTION_BINDING_VERSION: u32 = 1; diff --git a/distributed_cli/src/client_compiler/manifest/projections.rs b/distributed_cli/src/client_compiler/manifest/projections.rs index 8f57977c9..2543405b1 100644 --- a/distributed_cli/src/client_compiler/manifest/projections.rs +++ b/distributed_cli/src/client_compiler/manifest/projections.rs @@ -985,9 +985,7 @@ fn input_codec_compatible( (type_name, codec), ("Int", "int32") | ("BigInt", "json_number_precision_limited") ), - // Both numeric command codecs admit negative values. Without a - // non-negative refinement in the frozen manifest, neither proves U64. - ManifestProjectionValueType::U64 => false, + ManifestProjectionValueType::U64 => type_name == "BigInt" && unsigned_codec(codec), ManifestProjectionValueType::F64 => type_name == "Float" && codec == "float64", ManifestProjectionValueType::String => matches!( (type_name, codec), @@ -1004,7 +1002,7 @@ fn codec_compatible(codec: &str, expected: &ManifestProjectionValueType) -> bool ManifestProjectionValueType::I64 => { matches!(codec, "int32" | "json_number_precision_limited") } - ManifestProjectionValueType::U64 => false, + ManifestProjectionValueType::U64 => unsigned_codec(codec), ManifestProjectionValueType::F64 => codec == "float64", ManifestProjectionValueType::String | ManifestProjectionValueType::Enum(_) => { codec == "string" @@ -1013,6 +1011,10 @@ fn codec_compatible(codec: &str, expected: &ManifestProjectionValueType) -> bool } } +fn unsigned_codec(codec: &str) -> bool { + matches!(codec, "uint8" | "uint16" | "uint32" | "uint64_safe_integer") +} + fn constant_compatible( value: &ManifestProjectionValue, expected: &ManifestProjectionValueType, diff --git a/distributed_cli/src/client_compiler/render/commands.rs b/distributed_cli/src/client_compiler/render/commands.rs index 2de3d690e..d1657f434 100644 --- a/distributed_cli/src/client_compiler/render/commands.rs +++ b/distributed_cli/src/client_compiler/render/commands.rs @@ -428,7 +428,15 @@ fn render_command_field_type( } match field.codec.as_deref() { Some("boolean") => Ok("boolean".into()), - Some("float64" | "int32" | "json_number_precision_limited") => Ok("number".into()), + Some( + "float64" + | "int32" + | "json_number_precision_limited" + | "uint8" + | "uint16" + | "uint32" + | "uint64_safe_integer", + ) => Ok("number".into()), Some("string" | "base64" | "string_unvalidated_timestamp") => Ok("string".into()), Some("json") => Ok("ReplicaValue".into()), Some(codec) => Err(ClientCompileError::manifest( diff --git a/distributed_cli/src/client_compiler/tests.rs b/distributed_cli/src/client_compiler/tests.rs index 52dd91697..680458e6c 100644 --- a/distributed_cli/src/client_compiler/tests.rs +++ b/distributed_cli/src/client_compiler/tests.rs @@ -266,7 +266,7 @@ pub(super) fn manifest() -> JsonValue { "service_id": "todos-service", "surface": {"kind": "role", "name": "user"}, "schema_fingerprint": fingerprint("schema"), - "protocol_fingerprint": "sha256:00fb342f3acb4dc1c1716a43cc3001c748d5f6c500ff831690d820e9e43e2782", + "protocol_fingerprint": "sha256:0dfa8a3f49e17d8d99c5c095c1ed14f528cae3e55fbc2f2b852975a50936ec5b", "execution": { "max_depth": 8, "max_complexity": 500, @@ -3166,6 +3166,58 @@ fn command_protocol_and_extensions_are_preserved_exactly() { invalid_u64_source["projection_programs"][0]["arms"][0]["operations"][0]["fields"][1] ["assignment"]["expression"]["value_type"] = json!({"type": "u64"}); refresh_schema_fingerprint(&mut invalid_u64_source); + for codec in ["uint8", "uint16", "uint32", "uint64_safe_integer"] { + let mut unsigned = invalid_u64_source.clone(); + unsigned["commands"][0]["input"]["definition"]["fields"][1]["type_name"] = json!("BigInt"); + unsigned["commands"][0]["input"]["definition"]["fields"][1]["codec"] = json!(codec); + unsigned["models"][0]["fields"][2]["scalar"] = json!("BigInt"); + unsigned["models"][0]["fields"][2]["codec"] = json!("json_number_precision_limited"); + refresh_schema_fingerprint(&mut unsigned); + let compiled = compile_client(input_with_manifest( + unsigned.clone(), + "query Todos { todos { id priority } }", + )) + .expect("unsigned refined inputs prove U64 slots"); + let source = file(&compiled, "commands.ts"); + assert!(source.contains("readonly \"priority\": number;")); + let body = source + .split("export const Command_createTodo:") + .nth(1) + .unwrap() + .split_once(" = ") + .unwrap() + .1 + .split_once("\n};") + .unwrap() + .0; + let artifact: JsonValue = serde_json::from_str(&format!("{body}\n}}")).unwrap(); + assert_eq!(artifact["input"]["definition"]["fields"][1]["codec"], codec); + assert_eq!( + artifact["projection"]["preview"]["operations"][0]["mutation"]["fields"][1]["value"] + ["kind"], + "input" + ); + if codec == "uint64_safe_integer" { + let fixture: JsonValue = serde_json::from_str(include_str!( + "../../tests/fixtures/generated-unsigned-command.json" + )) + .unwrap(); + assert_eq!( + artifact, fixture, + "unsigned JS bridge must match generated output" + ); + } + unsigned["commands"][0]["extensions"]["trusted_presets"] = + json!([{"name": "priority", "codec": codec}]); + unsigned["commands"][0]["extensions"]["projection"]["preview_occurrences"][0]["values"] + [2]["source"] = json!({"kind": "trusted_preset", "name": "priority", "codec": codec}); + refresh_schema_fingerprint(&mut unsigned); + compile_client(input_with_manifest( + unsigned, + "query Todos { todos { id priority } }", + )) + .expect("unsigned trusted preset codec proves U64 slot"); + } let mut invalid_constant_source = value.clone(); invalid_constant_source["commands"][0]["extensions"]["projection"]["preview_occurrences"][0] ["values"][2]["source"] = diff --git a/distributed_cli/tests/cli_client.rs b/distributed_cli/tests/cli_client.rs index 53d29d906..72b64f5ad 100644 --- a/distributed_cli/tests/cli_client.rs +++ b/distributed_cli/tests/cli_client.rs @@ -17,7 +17,7 @@ const ROLE_MANIFEST: &str = r#"{ "name": "user" }, "schema_fingerprint": "sha256:758a97e4f7e1e538e8be86d24abd3d50a8da2d5813d29abd7a04bfa092d05189", - "protocol_fingerprint": "sha256:00fb342f3acb4dc1c1716a43cc3001c748d5f6c500ff831690d820e9e43e2782", + "protocol_fingerprint": "sha256:0dfa8a3f49e17d8d99c5c095c1ed14f528cae3e55fbc2f2b852975a50936ec5b", "execution": { "max_depth": 8, "max_complexity": 500, diff --git a/distributed_cli/tests/cli_manifest.rs b/distributed_cli/tests/cli_manifest.rs index 86106da2d..90afb309d 100644 --- a/distributed_cli/tests/cli_manifest.rs +++ b/distributed_cli/tests/cli_manifest.rs @@ -78,7 +78,7 @@ fn client_manifest_uses_service_surface_export() { ); assert_eq!( manifest["protocol_fingerprint"], - "sha256:00fb342f3acb4dc1c1716a43cc3001c748d5f6c500ff831690d820e9e43e2782" + "sha256:0dfa8a3f49e17d8d99c5c095c1ed14f528cae3e55fbc2f2b852975a50936ec5b" ); assert_eq!(manifest["models"][0]["id"], "OrderView"); assert_eq!(manifest["models"][0]["record_revisions"], true); diff --git a/distributed_cli/tests/fixtures/generated-commands.ts b/distributed_cli/tests/fixtures/generated-commands.ts index 6994f8724..f7403b9a9 100644 --- a/distributed_cli/tests/fixtures/generated-commands.ts +++ b/distributed_cli/tests/fixtures/generated-commands.ts @@ -67,7 +67,7 @@ export const Command_importTodos: ReplicaCommandArtifact, + version_arms: &[TokenStream2], replay_arms: &[TokenStream2], upcasters_method: &TokenStream2, ) -> TokenStream2 { @@ -43,6 +44,18 @@ pub(crate) fn aggregate_impl_tokens( &mut self, event: &distributed::EventRecord, ) -> Result<(), Self::ReplayError> { + // Hydration runs upcasters first. Payload compatibility alone + // cannot establish that an event has the handler's semantics. + let expected_version: u64 = match event.event_name.as_str() { + #(#version_arms)* + _ => return Err(format!("Unknown event: {}", event.event_name)), + }; + if event.event_version != expected_version { + return Err(format!( + "Unsupported event version for {}: expected {}, got {}", + event.event_name, expected_version, event.event_version, + )); + } match event.event_name.as_str() { #(#replay_arms)* _ => return Err(format!("Unknown event: {}", event.event_name)), @@ -178,6 +191,15 @@ pub(crate) fn expand_aggregate(input: TokenStream2) -> syn::Result let agg_name = &input.agg_name; let entity_field = &input.entity_field; + let version_arms: Vec<_> = input + .events + .iter() + .map(|event| { + let name = &event.event_name; + let version = &event.version; + quote! { #name => #version, } + }) + .collect(); // Generate replay match arms - deserialize and call method directly let replay_arms: Vec<_> = input @@ -249,6 +271,7 @@ pub(crate) fn expand_aggregate(input: TokenStream2) -> syn::Result agg_name, entity_field, &aggregate_type_method, + &version_arms, &replay_arms, &upcasters_method, ); @@ -327,6 +350,7 @@ struct AggregateInput { struct EventDef { event_name: LitStr, + version: syn::LitInt, args: Vec, method_name: Ident, method_args: Option>, // None = use event args, Some([]) = no args, Some([x,y]) = specific args @@ -382,6 +406,19 @@ impl Parse for AggregateInput { args_content.parse_terminated(Ident::parse, Token![,])?; let args: Vec = args.into_iter().collect(); + // `"renamed"(name), version = 2 => rename`; omitted versions + // have the same v1 default as #[digest] and #[event]. + let version = if content.peek(Token![,]) { + content.parse::()?; + let keyword: Ident = content.parse()?; + if keyword != "version" { + return Err(syn::Error::new(keyword.span(), "expected `version`")); + } + content.parse::()?; + content.parse::()? + } else { + syn::LitInt::new("1", event_name.span()) + }; content.parse::]>()?; let method_name: Ident = content.parse()?; @@ -398,6 +435,7 @@ impl Parse for AggregateInput { events.push(EventDef { event_name, + version, args, method_name, method_args, diff --git a/distributed_macros/src/command_types.rs b/distributed_macros/src/command_types.rs index c67077002..343e8a79e 100644 --- a/distributed_macros/src/command_types.rs +++ b/distributed_macros/src/command_types.rs @@ -167,6 +167,7 @@ fn expand( }); let (type_name, nullable, list, item_nullable, nested) = map_type(&field.ty, field, &nested_trait, &method)?; + let unsigned_integer = unsigned_integer_tokens(&field.ty, &framework); let effect_path_kind = if !list && nested.is_some() { quote! { #framework::command::EffectInputObjectKind } } else { @@ -184,6 +185,7 @@ fn expand( nullable: #nullable, list: #list, item_nullable: #item_nullable, + unsigned_integer: #unsigned_integer, nested: #nested_tokens, } }); @@ -342,6 +344,38 @@ fn map_type( Ok((ident, nullable, list, item_nullable, Some(nested))) } +fn unsigned_integer_tokens(ty: &Type, framework: &TokenStream) -> TokenStream { + let mut current = ty; + while let Some(inner) = + extract_path_arg(current, "Option").or_else(|| extract_path_arg(current, "Vec")) + { + current = inner; + } + let Type::Path(path) = current else { + return quote! { None }; + }; + match path + .path + .segments + .last() + .map(|segment| segment.ident.to_string()) + .as_deref() + { + Some("u8") => quote! { Some(#framework::command::CommandUnsignedInteger::U8) }, + Some("u16") => quote! { Some(#framework::command::CommandUnsignedInteger::U16) }, + Some("u32") => quote! { Some(#framework::command::CommandUnsignedInteger::U32) }, + Some("u64") => quote! { Some(#framework::command::CommandUnsignedInteger::U64) }, + Some("usize") => quote! { + Some(match usize::BITS { + 16 => #framework::command::CommandUnsignedInteger::U16, + 32 => #framework::command::CommandUnsignedInteger::U32, + _ => #framework::command::CommandUnsignedInteger::U64, + }) + }, + _ => quote! { None }, + } +} + fn extract_path_arg<'a>(ty: &'a Type, wrapper: &str) -> Option<&'a Type> { let Type::Path(path) = ty else { return None; diff --git a/distributed_macros/src/sourced.rs b/distributed_macros/src/sourced.rs index 90caa3434..b96f63d57 100644 --- a/distributed_macros/src/sourced.rs +++ b/distributed_macros/src/sourced.rs @@ -212,6 +212,7 @@ fn find_and_remove_event_attr( struct EventMethodInfo { event_name: LitStr, + version: syn::LitInt, method_name: Ident, params: Vec<(Ident, syn::Type)>, /// Present when this recorder has `domain` and therefore a generated @@ -833,6 +834,7 @@ pub(crate) fn expand_sourced(attr: TokenStream2, item: TokenStream2) -> syn::Res }; event_methods.push(EventMethodInfo { + version: event_version(event_attr.version.as_ref()), event_name: event_attr.event_name, method_name: method.sig.ident.clone(), params, @@ -977,6 +979,14 @@ pub(crate) fn expand_sourced(attr: TokenStream2, item: TokenStream2) -> syn::Res // Generate impl Aggregate let entity_field = &args.entity_field; + let version_arms: Vec<_> = event_methods + .iter() + .map(|event| { + let name = &event.event_name; + let version = &event.version; + quote! { #name => #version, } + }) + .collect(); let replay_arms: Vec<_> = event_methods .iter() .map(|e| { @@ -1023,6 +1033,7 @@ pub(crate) fn expand_sourced(attr: TokenStream2, item: TokenStream2) -> syn::Res &struct_name, entity_field, &aggregate_type_method, + &version_arms, &replay_arms, &upcasters_method, ); diff --git a/distributed_macros/tests/compile_fail/application_command_duplicate_id.rs b/distributed_macros/tests/compile_fail/application_command_duplicate_id.rs index e173a3023..39b8e47b2 100644 --- a/distributed_macros/tests/compile_fail/application_command_duplicate_id.rs +++ b/distributed_macros/tests/compile_fail/application_command_duplicate_id.rs @@ -10,6 +10,7 @@ fn spec() -> CommandSpec { CommandTypeSpec { name: "DuplicateInput".into(), fields: vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/distributed_macros/tests/compile_fail/application_command_duplicate_id.stderr b/distributed_macros/tests/compile_fail/application_command_duplicate_id.stderr index 5377eae50..a6b983c15 100644 --- a/distributed_macros/tests/compile_fail/application_command_duplicate_id.stderr +++ b/distributed_macros/tests/compile_fail/application_command_duplicate_id.stderr @@ -1,12 +1,12 @@ error[E0080]: evaluation panicked: duplicate command identity in module declaration - --> tests/compile_fail/application_command_duplicate_id.rs:38:1 + --> tests/compile_fail/application_command_duplicate_id.rs:39:1 | -38 | / distributed::module! { -39 | | pub DUPLICATE_MODULE { -40 | | id: "duplicates", -41 | | commands: [FIRST_DEFINITION, SECOND_DEFINITION], -42 | | } -43 | | } +39 | / distributed::module! { +40 | | pub DUPLICATE_MODULE { +41 | | id: "duplicates", +42 | | commands: [FIRST_DEFINITION, SECOND_DEFINITION], +43 | | } +44 | | } | |_^ evaluation of `_` failed inside this call | note: inside `assert_unique_command_ids` diff --git a/docs/unsigned-command-inputs.md b/docs/unsigned-command-inputs.md new file mode 100644 index 000000000..ae183f9ae --- /dev/null +++ b/docs/unsigned-command-inputs.md @@ -0,0 +1,55 @@ +# Unsigned command inputs and optimistic projections + +Use the Rust type that expresses the domain boundary: + +```rust,ignore +#[derive(serde::Deserialize, distributed::CommandInput)] +pub struct UpdateDocumentInput { + pub document_id: String, + pub expected_revision: u64, +} +``` + +`CommandInput` and `CommandOutput` retain unsigned width automatically, including +optional fields and list elements. A field remains GraphQL `BigInt` and TypeScript +`number`. Its generated command codec carries the narrower unsigned contract: + +| Rust type | Command codec | Browser range | +| --- | --- | --- | +| `u8` | `uint8` | 0–255 | +| `u16` | `uint16` | 0–65,535 | +| `u32` | `uint32` | 0–4,294,967,295 | +| `u64` | `uint64_safe_integer` | 0–9,007,199,254,740,991 | + +`usize` follows the target's pointer width. Fixed-width types are preferable +for portable domain contracts. + +For example, an event preview can bind `input.expected_revision` to a typed +`u64` event field. The manifest compiler now proves that assignment from the +unsigned codec, and the generated command prepares the same integer in its +optimistic projection and transport input. No browser annotation or handwritten +optimistic mutation is needed. + +Command preparation rejects negative numbers, negative zero, fractions, +non-finite numbers, strings, and numbers above the codec's maximum before any +optimistic effect or dispatch. The same validation applies to generated command +results and trusted presets. Explicit trusted presets targeting U64 projection +slots must declare an unsigned codec; an unrestricted signed `BigInt` preset +cannot establish that proof. + +JavaScript numbers cannot represent every `u64` exactly. The browser boundary +deliberately stops at `Number.MAX_SAFE_INTEGER`. Rust command execution retains +the full unsigned range, and projection constants and stored aggregate state +retain their existing full-width U64 representation. Signed command codecs and +SQL read-model scalar codecs are unchanged. + +## Regeneration + +This changes command contract and protocol fingerprints. Rebuild the service +and generated clients together with `distributed build` or `distributed dev`. +Mixed old/new artifacts fail the existing fingerprint check. + +Code that manually constructs `CommandTypeField` or `SurfaceTypeField` must +set `unsigned_integer` to the matching `CommandUnsignedInteger` variant, or +`None` for unrefined fields. Prefer derives so this metadata follows the Rust +type. A refinement on a non-`BigInt` field is rejected. diff --git a/js/src/protocol.ts b/js/src/protocol.ts index d6b3dd457..0cbcaac18 100644 --- a/js/src/protocol.ts +++ b/js/src/protocol.ts @@ -14,6 +14,7 @@ import { parseDistributedGenerationEnvelope, type DistributedGenerationEnvelope } from './generation.js'; +import { isUnsignedInteger, unsignedIntegerMaximum, type UnsignedIntegerCodec } from './unsigned-integer.js'; /** The only Distributed GraphQL protocol version understood by this package. */ export const DISTRIBUTED_PROTOCOL_VERSION = 1 as const; @@ -51,6 +52,7 @@ export type DistributedProjectionDisposition = 'revalidate'; /** Closed wire codecs for server-derived, client-visible trusted presets. */ export type DistributedTrustedPresetCodec = + | UnsignedIntegerCodec | 'string' | 'string_unvalidated_timestamp' | 'base64' @@ -828,6 +830,7 @@ export function isDistributedTrustedPresetCodec( value === 'int32' || value === 'float64' || value === 'json_number_precision_limited' || + (typeof value === 'string' && unsignedIntegerMaximum(value) !== undefined) || value === 'json' ); } @@ -845,6 +848,11 @@ function parseTrustedPresetValue( codec: DistributedTrustedPresetCodec, path: string ): DistributedProtocolValue { + const unsignedMaximum = unsignedIntegerMaximum(codec); + if (unsignedMaximum !== undefined) { + if (!isUnsignedInteger(value, unsignedMaximum)) invalid(path); + return value; + } switch (codec) { case 'string': case 'string_unvalidated_timestamp': @@ -887,6 +895,8 @@ function parseTrustedPresetValue( return Object.is(value, -0) ? 0 : value; case 'json': return cloneProtocolValue(value, path); + default: + return invalid(path); } } diff --git a/js/src/replica/command-runtime/lib/output.ts b/js/src/replica/command-runtime/lib/output.ts index 7929cd371..2f5c53099 100644 --- a/js/src/replica/command-runtime/lib/output.ts +++ b/js/src/replica/command-runtime/lib/output.ts @@ -13,6 +13,7 @@ import { import { ReplicaCommandRuntimeError } from '../errors.js'; import { compareCodeUnits } from '../../../lib/compare-code-units.js'; import { isPlainRecord } from '../../../lib/is-plain-record.js'; +import { isUnsignedInteger, unsignedIntegerMaximum } from '../../../unsigned-integer.js'; import { comparePropertyKeys, outputInvalid @@ -164,6 +165,11 @@ export function cloneOutputScalar( value: unknown, path: string ): ReplicaValue { + const unsignedMaximum = unsignedIntegerMaximum(codec); + if (unsignedMaximum !== undefined) { + if (!isUnsignedInteger(value, unsignedMaximum)) outputInvalid(path); + return value; + } switch (codec) { case 'string': case 'string_unvalidated_timestamp': @@ -251,4 +257,3 @@ export function cloneOutputJson( active.delete(value); return Object.freeze(output); } - diff --git a/js/src/replica/commands/clone.ts b/js/src/replica/commands/clone.ts index f6d1eb916..6c9230c0b 100644 --- a/js/src/replica/commands/clone.ts +++ b/js/src/replica/commands/clone.ts @@ -1,4 +1,5 @@ import type { ReplicaClientSurface, ReplicaValue } from '../types.js'; +import { isUnsignedInteger, unsignedIntegerMaximum } from '../../unsigned-integer.js'; import { createReplicaCommandId } from '../command-id.js'; import { isPlainRecord } from '../../lib/is-plain-record.js'; import { @@ -169,6 +170,11 @@ export function cloneScalar( value: unknown, path: string ): ReplicaValue { + const unsignedMaximum = unsignedIntegerMaximum(codec); + if (unsignedMaximum !== undefined) { + if (!isUnsignedInteger(value, unsignedMaximum)) inputInvalid(path); + return value; + } switch (codec) { case 'string': case 'string_unvalidated_timestamp': diff --git a/js/src/unsigned-integer.ts b/js/src/unsigned-integer.ts new file mode 100644 index 000000000..ec64f00ea --- /dev/null +++ b/js/src/unsigned-integer.ts @@ -0,0 +1,17 @@ +/** Exact browser-number refinements of Rust unsigned command integers. */ +export type UnsignedIntegerCodec = 'uint8' | 'uint16' | 'uint32' | 'uint64_safe_integer'; + +export function unsignedIntegerMaximum(codec: string | undefined): number | undefined { + switch (codec) { + case 'uint8': return 255; + case 'uint16': return 65_535; + case 'uint32': return 4_294_967_295; + case 'uint64_safe_integer': return Number.MAX_SAFE_INTEGER; + default: return undefined; + } +} + +export function isUnsignedInteger(value: unknown, maximum: number): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) + && !Object.is(value, -0) && value >= 0 && value <= maximum; +} diff --git a/js/tests/protocol-transport.test.mjs b/js/tests/protocol-transport.test.mjs index 35542556d..16ea9f5bc 100644 --- a/js/tests/protocol-transport.test.mjs +++ b/js/tests/protocol-transport.test.mjs @@ -103,6 +103,22 @@ function responseExtensions() { }; } +test('unsigned trusted presets accept only exact bounded integers', () => { + for (const [codec, maximum] of [['uint8', 255], ['uint16', 65_535], + ['uint32', 4_294_967_295], ['uint64_safe_integer', Number.MAX_SAFE_INTEGER]]) { + for (const value of [0, 1, maximum]) { + const envelope = distributedEnvelope(); + envelope.trustedPresets = [{ name: 'revision', codec, value }]; + assert.equal(parseDistributedProtocolEnvelope(envelope).trustedPresets[0].value, value); + } + for (const value of [-1, -0, 1.5, NaN, Infinity, maximum + 1, '1', 1n]) { + const envelope = distributedEnvelope(); + envelope.trustedPresets = [{ name: 'revision', codec, value }]; + assert.throws(() => parseDistributedProtocolEnvelope(envelope), DistributedProtocolError); + } + } +}); + test('protocol parser validates receipts while retaining future metadata opaquely', () => { assert.deepEqual(COMMAND_CONSISTENCY, { SUCCEEDED: 'succeeded', diff --git a/js/tests/replica-command-artifacts.test.mjs b/js/tests/replica-command-artifacts.test.mjs index 0b7abd29b..c01e3d381 100644 --- a/js/tests/replica-command-artifacts.test.mjs +++ b/js/tests/replica-command-artifacts.test.mjs @@ -32,6 +32,38 @@ const GENERATED_DRAINING_COMMAND = JSON.parse( ) ); +const GENERATED_UNSIGNED_COMMAND = JSON.parse(readFileSync(new URL( + '../../distributed_cli/tests/fixtures/generated-unsigned-command.json', import.meta.url +), 'utf8')); + +test('generated unsigned command preserves exact optimistic and transport values', () => { + for (const priority of [0, 1, Number.MAX_SAFE_INTEGER]) { + const prepared = prepareReplicaCommand(GENERATED_UNSIGNED_COMMAND, { + id: GENERATED_UUID, tenantId: 'tenant-1', title: 'Update', priority + }, { commandId: COMMAND_ID }); + assert.equal(prepared.transport.variables.input.priority, priority); + assert.equal(prepared.optimistic.operations[0].fields.priority, priority); + assert.equal(prepared.optimistic.operations[0].kind, 'upsert'); + } +}); + +test('unsigned command preparation rejects invalid values before producing optimism', () => { + for (const [codec, maximum] of [['uint8', 255], ['uint16', 65_535], + ['uint32', 4_294_967_295], ['uint64_safe_integer', Number.MAX_SAFE_INTEGER]]) { + const artifact = structuredClone(GENERATED_UNSIGNED_COMMAND); + artifact.input.definition.fields.find(field => field.name === 'priority').codec = codec; + const prepare = priority => prepareReplicaCommand(artifact, { + id: GENERATED_UUID, tenantId: 'tenant-1', title: 'Update', priority + }, { commandId: COMMAND_ID }); + assert.equal(prepare(maximum).optimistic.operations[0].fields.priority, maximum); + for (const invalid of [-1, -0, 0.5, NaN, Infinity, -Infinity, maximum + 1, + 2 ** 64, '1', 1n, null, undefined]) { + assert.throws(() => prepare(invalid), error => + error instanceof ReplicaCommandContractError && /input.priority/.test(error.message)); + } + } +}); + test('compiled recovery-only command drops unused presets without weakening validation', () => { const artifact = JSON.parse(readFileSync(new URL( '../../distributed_cli/tests/fixtures/generated-recovery-preset-command.json', import.meta.url diff --git a/js/tests/unsigned-command-output.test.mjs b/js/tests/unsigned-command-output.test.mjs new file mode 100644 index 000000000..357dec36b --- /dev/null +++ b/js/tests/unsigned-command-output.test.mjs @@ -0,0 +1,15 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { cloneOutputScalar } from '../dist/replica/command-runtime/lib/output.js'; + +test('unsigned command results reject lossy numbers before entering the replica', () => { + for (const [codec, maximum] of [['uint8', 255], ['uint16', 65_535], + ['uint32', 4_294_967_295], ['uint64_safe_integer', Number.MAX_SAFE_INTEGER]]) { + assert.equal(cloneOutputScalar(codec, maximum, 'result.revision'), maximum); + for (const value of [-1, -0, 0.5, NaN, Infinity, maximum + 1, 2 ** 64, '1']) { + assert.throws(() => cloneOutputScalar(codec, value, 'result.revision')); + } + } + assert.equal(cloneOutputScalar('json_number_precision_limited', -1, 'signed'), -1); + assert.equal(cloneOutputScalar('int32', -2_147_483_648, 'signed'), -2_147_483_648); +}); diff --git a/src/application/command.rs b/src/application/command.rs index f3a604c11..b1651e42e 100644 --- a/src/application/command.rs +++ b/src/application/command.rs @@ -22,6 +22,8 @@ pub struct CommandTypeField { pub list: bool, pub item_nullable: bool, #[serde(default, skip_serializing_if = "Option::is_none")] + pub unsigned_integer: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub nested: Option>, } @@ -43,6 +45,7 @@ impl From<&CommandTypeDef> for CommandTypeSpec { .fields .iter() .map(|field| CommandTypeField { + unsigned_integer: field.unsigned_integer, name: field.name.clone(), type_name: field.type_name.clone(), nullable: field.nullable, diff --git a/src/application/module.rs b/src/application/module.rs index e7122991f..d47d81848 100644 --- a/src/application/module.rs +++ b/src/application/module.rs @@ -664,6 +664,7 @@ fn surface_type_spec(definition: &SurfaceTypeDef) -> CommandTypeSpec { .fields .iter() .map(|field| super::command::CommandTypeField { + unsigned_integer: field.unsigned_integer, name: field.name.clone(), type_name: field.type_name.clone(), nullable: field.nullable, diff --git a/src/command/input.rs b/src/command/input.rs index f8fc5bc7e..94cc8ebe5 100644 --- a/src/command/input.rs +++ b/src/command/input.rs @@ -221,6 +221,23 @@ fn canonicalize_leaf( value: Value, path: &str, ) -> Result { + if let Some(unsigned) = field.unsigned_integer { + if field.type_name != "BigInt" || field.nested.is_some() { + return Err(CommandInputError::at( + path, + "unsigned refinement requires BigInt", + )); + } + return match value { + Value::Number(value) if value.as_u64().is_some_and(|n| n <= unsigned.max_value()) => { + Ok(Value::Number(value)) + } + _ => Err(CommandInputError::at( + path, + "must be an unsigned integer in range", + )), + }; + } if let Some(nested) = field.nested.as_deref() { return canonicalize_object(nested, value, path); } @@ -401,6 +418,7 @@ mod tests { nested: Option, ) -> CommandTypeField { CommandTypeField { + unsigned_integer: None, name: name.into(), type_name: type_name.into(), nullable, @@ -433,6 +451,44 @@ mod tests { ) } + #[test] + fn unsigned_inputs_preserve_full_rust_range_and_reject_invalid_values() { + use crate::command::CommandUnsignedInteger; + for unsigned in [ + CommandUnsignedInteger::U8, + CommandUnsignedInteger::U16, + CommandUnsignedInteger::U32, + CommandUnsignedInteger::U64, + ] { + let mut value_field = field("revision", "BigInt", false, false, false, None); + value_field.unsigned_integer = Some(unsigned); + let definition = CommandTypeDef::new("RevisionInput", vec![value_field]); + for value in [0, 1, unsigned.max_value()] { + assert_eq!( + canonicalize_command_input(&definition, json!({"revision": value})) + .unwrap() + .wire()["revision"], + json!(value) + ); + } + let mut invalid = vec![ + json!(-1), + json!(-0.0), + json!(1.5), + json!("1"), + serde_json::from_str("18446744073709551616").unwrap(), + ]; + if let Some(overflow) = unsigned.max_value().checked_add(1) { + invalid.push(json!(overflow)); + } + for value in invalid { + assert!( + canonicalize_command_input(&definition, json!({"revision": value})).is_err() + ); + } + } + } + #[test] fn key_order_is_canonical_but_lists_are_preserved() { let left = canonicalize_command_input( diff --git a/src/command/mod.rs b/src/command/mod.rs index 991e6beff..47915df88 100644 --- a/src/command/mod.rs +++ b/src/command/mod.rs @@ -12,7 +12,9 @@ pub(crate) mod input; mod types; pub(crate) use types::scalar_type_name; -pub use types::{CommandInputType, CommandOutputType, CommandTypeDef, CommandTypeField}; +pub use types::{ + CommandInputType, CommandOutputType, CommandTypeDef, CommandTypeField, CommandUnsignedInteger, +}; mod direct_projection; mod effect_wire; diff --git a/src/command/tests.rs b/src/command/tests.rs index 9ae7d2eb9..e80dc9c96 100644 --- a/src/command/tests.rs +++ b/src/command/tests.rs @@ -22,6 +22,18 @@ fn command_ledger_fingerprint_preserves_the_v1_canonical_contract() { ); } +#[test] +fn unsigned_range_is_part_of_command_retry_identity() { + let mut signed = typed_command::>("revision.update").into_contract(); + signed.input.fields[0].type_name = "BigInt".into(); + let mut unsigned = signed.clone(); + unsigned.input.fields[0].unsigned_integer = Some(CommandUnsignedInteger::U64); + assert_ne!(signed.fingerprint_bytes(), unsigned.fingerprint_bytes()); + let mut narrower = unsigned.clone(); + narrower.input.fields[0].unsigned_integer = Some(CommandUnsignedInteger::U32); + assert_ne!(unsigned.fingerprint_bytes(), narrower.fingerprint_bytes()); +} + #[allow(dead_code)] #[derive(Deserialize)] struct Input { @@ -33,6 +45,7 @@ impl CommandInputType for Input { CommandTypeDef::new( "Input", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -141,6 +154,7 @@ impl CommandOutputType for Payload { CommandTypeDef::new( "Payload", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -873,6 +887,7 @@ fn binding_canonicalizes_fields_and_roles_but_preserves_effect_order() { .roles(["writer", "admin"]) .into_contract(); first.input.fields.push(CommandTypeField { + unsigned_integer: None, name: "z_extra".into(), type_name: "String".into(), nullable: true, @@ -900,6 +915,7 @@ fn binding_canonicalizes_fields_and_roles_but_preserves_effect_order() { .roles(["writer", "admin"]) .into_contract(); reordered.input.fields.push(CommandTypeField { + unsigned_integer: None, name: "z_extra".into(), type_name: "String".into(), nullable: true, diff --git a/src/command/typed_command.rs b/src/command/typed_command.rs index 221b0ffe8..a51e1db66 100644 --- a/src/command/typed_command.rs +++ b/src/command/typed_command.rs @@ -347,14 +347,20 @@ fn canonical_command_type(definition: &CommandTypeDef) -> serde_json::Value { fields.sort_by(|left, right| left.name.cmp(&right.name)); serde_json::json!({ "name": definition.name, - "fields": fields.into_iter().map(|field| serde_json::json!({ + "fields": fields.into_iter().map(|field| { + let mut value = serde_json::json!({ "name": field.name, "type_name": field.type_name, "nullable": field.nullable, "list": field.list, "item_nullable": field.item_nullable, "nested": field.nested.as_deref().map(canonical_command_type), - })).collect::>(), + }); + if let Some(unsigned) = field.unsigned_integer { + value["unsigned_integer"] = serde_json::json!(unsigned); + } + value + }).collect::>(), }) } diff --git a/src/command/types.rs b/src/command/types.rs index 4e38cc946..7f055aefc 100644 --- a/src/command/types.rs +++ b/src/command/types.rs @@ -9,6 +9,51 @@ use std::any::TypeId; use crate::read_model::RelationalReadModel; use crate::table::ColumnType; +/// Rust unsigned integer range retained by command contracts and adapters. +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CommandUnsignedInteger { + U8, + U16, + U32, + U64, +} + +impl CommandUnsignedInteger { + pub fn max_value(self) -> u64 { + match self { + Self::U8 => u64::from(u8::MAX), + Self::U16 => u64::from(u16::MAX), + Self::U32 => u64::from(u32::MAX), + Self::U64 => u64::MAX, + } + } + + /// Browser numbers must be exact; U64 therefore stops at 2^53 - 1. + pub fn client_codec(self) -> &'static str { + match self { + Self::U8 => "uint8", + Self::U16 => "uint16", + Self::U32 => "uint32", + Self::U64 => "uint64_safe_integer", + } + } + + pub fn from_client_codec(codec: &str) -> Option { + match codec { + "uint8" => Some(Self::U8), + "uint16" => Some(Self::U16), + "uint32" => Some(Self::U32), + "uint64_safe_integer" => Some(Self::U64), + _ => None, + } + } + + pub fn client_max_value(self) -> u64 { + self.max_value().min(9_007_199_254_740_991) + } +} + /// One field on a command input or output object. #[derive(Clone, Debug, PartialEq, Eq)] pub struct CommandTypeField { @@ -18,6 +63,8 @@ pub struct CommandTypeField { pub list: bool, /// Whether list elements are nullable. Always `false` for non-list fields. pub item_nullable: bool, + /// A typed refinement of BigInt, including list elements when `list` is true. + pub unsigned_integer: Option, /// Nested object type definition when `type_name` is not a scalar. pub nested: Option>, } @@ -107,6 +154,8 @@ where ) }); CommandTypeField { + // Relational result codecs follow the read-model wire contract. + unsigned_integer: None, name: column.column_name.clone(), type_name: type_name.into(), nullable: column.nullable, diff --git a/src/graphql/client_manifest/codec.rs b/src/graphql/client_manifest/codec.rs index 957b0d6b6..dcd1bd2db 100644 --- a/src/graphql/client_manifest/codec.rs +++ b/src/graphql/client_manifest/codec.rs @@ -184,7 +184,17 @@ fn client_type(definition: &SurfaceTypeDef) -> Result Result { projection_binding_version: u32, projection_operation_semantics_version: u32, command_projection_extension_version: u32, + unsigned_command_codecs_version: u32, scalar_codecs: Vec, } hash_json(&ProtocolMaterial { @@ -275,6 +286,7 @@ pub(super) fn protocol_fingerprint() -> Result { super::projections::CLIENT_PROJECTION_OPERATION_SEMANTICS_VERSION, command_projection_extension_version: super::projections::COMMAND_PROJECTION_EXTENSION_VERSION, + unsigned_command_codecs_version: 1, scalar_codecs: supported_scalar_codecs(), }) } diff --git a/src/graphql/client_manifest/projections.rs b/src/graphql/client_manifest/projections.rs index bc79d5025..61f4284d8 100644 --- a/src/graphql/client_manifest/projections.rs +++ b/src/graphql/client_manifest/projections.rs @@ -1189,9 +1189,10 @@ fn input_field_compatible( ) -> bool { match expected { ProjectionValueType::Boolean => field.type_name == "Boolean", - ProjectionValueType::I64 | ProjectionValueType::U64 => { - matches!(field.type_name.as_str(), "BigInt" | "Int") + ProjectionValueType::I64 => { + matches!(field.type_name.as_str(), "BigInt" | "Int") && field.unsigned_integer.is_none() } + ProjectionValueType::U64 => field.type_name == "BigInt" && field.unsigned_integer.is_some(), ProjectionValueType::F64 => field.type_name == "Float", ProjectionValueType::String => { matches!(field.type_name.as_str(), "ID" | "String" | "Timestamptz") @@ -1211,8 +1212,11 @@ fn input_field_compatible( fn codec_compatible(codec: &str, expected: &ProjectionValueType) -> bool { match expected { ProjectionValueType::Boolean => codec == "boolean", - ProjectionValueType::I64 | ProjectionValueType::U64 => { - codec == "json_number_precision_limited" + ProjectionValueType::I64 => { + matches!(codec, "int32" | "json_number_precision_limited") + } + ProjectionValueType::U64 => { + crate::command::CommandUnsignedInteger::from_client_codec(codec).is_some() } ProjectionValueType::F64 => codec == "float64", ProjectionValueType::String | ProjectionValueType::Enum(_) => codec == "string", @@ -1416,6 +1420,7 @@ mod tests { input: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "TodoPreviewInput".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "value".into(), type_name: input_type.into(), nullable: false, @@ -1730,6 +1735,7 @@ mod tests { name: "TodoCreateInput".into(), fields: vec![ SurfaceTypeField { + unsigned_integer: None, name: "todo_id".into(), type_name: "ID".into(), nullable: false, @@ -1738,6 +1744,7 @@ mod tests { nested: None, }, SurfaceTypeField { + unsigned_integer: None, name: "title".into(), type_name: "String".into(), nullable: false, @@ -2303,6 +2310,43 @@ mod tests { let string_command = command("String"); let input = CommandProjectionPreviewSource::input(["value"]); + let mut unsigned_command = command("BigInt"); + let signed_command = unsigned_command.clone(); + let SurfaceCommandShape::Typed(definition) = &mut unsigned_command.input else { + unreachable!() + }; + definition.fields[0].unsigned_integer = Some(crate::command::CommandUnsignedInteger::U64); + for (command, allowed) in [(&unsigned_command, true), (&signed_command, false)] { + let source = client_preview_source( + &input, + true, + None, + Some(ProjectionPortableType::U64), + Some("u64"), + Some(false), + Some(true), + &ProjectionValueType::U64, + command, + ) + .unwrap(); + assert_eq!( + matches!(source, ClientProjectionPreviewSource::Input { .. }), + allowed + ); + } + assert!(codec_compatible( + "uint64_safe_integer", + &ProjectionValueType::U64 + )); + assert!(!codec_compatible( + "json_number_precision_limited", + &ProjectionValueType::U64 + )); + assert!(codec_compatible( + "json_number_precision_limited", + &ProjectionValueType::I64 + )); + assert_eq!( client_preview_source( &input, diff --git a/src/graphql/client_manifest/tests.rs b/src/graphql/client_manifest/tests.rs index c0cb568ba..4f2e876d6 100644 --- a/src/graphql/client_manifest/tests.rs +++ b/src/graphql/client_manifest/tests.rs @@ -191,6 +191,7 @@ impl CommandInputType for CompleteInput { CommandTypeDef::new( "CompleteTodoInput", vec![CommandTypeField { + unsigned_integer: None, name: "todo_id".into(), type_name: "String".into(), nullable: false, @@ -210,6 +211,7 @@ impl CommandOutputType for CompletePayload { CommandTypeDef::new( "CompleteTodoPayload", vec![CommandTypeField { + unsigned_integer: None, name: "todo_id".into(), type_name: "String".into(), nullable: false, @@ -491,6 +493,7 @@ fn projected_surface() -> Surface { .columns .iter() .map(|column| SurfaceTypeField { + unsigned_integer: None, name: column.name.clone(), type_name: column.scalar.clone(), nullable: column.nullable, @@ -507,6 +510,7 @@ fn projected_surface() -> Surface { input: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "ProjectTodoInput".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "todo_id".into(), type_name: "String".into(), nullable: false, @@ -767,7 +771,7 @@ fn role_manifest_is_deterministic_and_hides_denied_identity_and_commands() { ); assert_eq!( first.protocol_fingerprint, - "sha256:00fb342f3acb4dc1c1716a43cc3001c748d5f6c500ff831690d820e9e43e2782" + "sha256:0dfa8a3f49e17d8d99c5c095c1ed14f528cae3e55fbc2f2b852975a50936ec5b" ); let user = first @@ -878,6 +882,33 @@ fn role_manifest_is_deterministic_and_hides_denied_identity_and_commands() { ); } +#[test] +fn unsigned_command_fields_export_refined_codecs_and_change_fingerprints() { + use crate::command::CommandUnsignedInteger; + let mut full = full_surface(); + let command = &mut full.commands[0]; + let SurfaceCommandShape::Typed(definition) = &mut command.input else { + unreachable!() + }; + definition.fields[0].type_name = "BigInt".into(); + let signed = manifest_for_all_models("todos-service", "user", &full); + let SurfaceCommandShape::Typed(definition) = &mut full.commands[0].input else { + unreachable!() + }; + definition.fields[0].unsigned_integer = Some(CommandUnsignedInteger::U64); + let unsigned = manifest_for_all_models("todos-service", "user", &full); + let wire = serde_json::to_value(&unsigned).unwrap(); + assert_eq!( + wire["commands"][0]["input"]["definition"]["fields"][0]["type_name"], + "BigInt" + ); + assert_eq!( + wire["commands"][0]["input"]["definition"]["fields"][0]["codec"], + "uint64_safe_integer" + ); + assert_ne!(signed.schema_fingerprint, unsigned.schema_fingerprint); +} + #[test] fn role_and_application_partition_manifests_hide_raw_paths_and_denied_values() { use crate::projection::placement::{ProjectionBindingState, ProjectionExecutionClass}; diff --git a/src/graphql/commands.rs b/src/graphql/commands.rs index 18d2ab970..c09c4c8db 100644 --- a/src/graphql/commands.rs +++ b/src/graphql/commands.rs @@ -31,6 +31,7 @@ fn surface_type(definition: &CommandTypeDef) -> SurfaceTypeDef { .fields .iter() .map(|field| SurfaceTypeField { + unsigned_integer: field.unsigned_integer, name: field.name.clone(), type_name: field.type_name.clone(), nullable: field.nullable, diff --git a/src/graphql/engine/protocol.rs b/src/graphql/engine/protocol.rs index 180577dee..36568159e 100644 --- a/src/graphql/engine/protocol.rs +++ b/src/graphql/engine/protocol.rs @@ -236,6 +236,14 @@ pub(crate) fn resolve_protocol_preset( } serde_json::Value::Number(parsed.into()) } + codec if crate::command::CommandUnsignedInteger::from_client_codec(codec).is_some() => { + let unsigned = crate::command::CommandUnsignedInteger::from_client_codec(codec)?; + let parsed = raw.parse::().ok()?; + if parsed > unsigned.client_max_value() || parsed.to_string() != raw { + return None; + } + serde_json::Value::Number(parsed.into()) + } "float64" => { let parsed = raw.parse::().ok()?; if !parsed.is_finite() { diff --git a/src/graphql/engine/tests.rs b/src/graphql/engine/tests.rs index ea2b2d2d9..8576815a4 100644 --- a/src/graphql/engine/tests.rs +++ b/src/graphql/engine/tests.rs @@ -943,6 +943,51 @@ mod client_surface_parity_tests { assert!(!response.extensions.contains_key("distributed")); } + #[test] + fn unsigned_presets_from_session_enforce_canonical_safe_ranges() { + for (codec, maximum) in [ + ("uint8", 255_u64), + ("uint16", 65_535), + ("uint32", 4_294_967_295), + ("uint64_safe_integer", 9_007_199_254_740_991), + ] { + let descriptor = ClientTrustedPresetDescriptor { + name: "x-revision".into(), + codec: codec.into(), + }; + let mut session = Session::new(); + for raw in ["0".to_string(), maximum.to_string()] { + session.set("x-revision", &raw); + assert_eq!( + crate::graphql::engine::protocol::resolve_protocol_preset( + &session, + &descriptor + ) + .unwrap() + .value, + serde_json::json!(raw.parse::().unwrap()) + ); + } + for raw in [ + "-1".to_string(), + "-0".into(), + "1.5".into(), + "1.0".into(), + "+1".into(), + "01".into(), + "NaN".into(), + (maximum + 1).to_string(), + ] { + session.set("x-revision", &raw); + assert!(crate::graphql::engine::protocol::resolve_protocol_preset( + &session, + &descriptor + ) + .is_none()); + } + } + } + #[cfg(feature = "sqlite")] #[tokio::test] async fn row_policy_presets_follow_sql_claim_case_normalization() { @@ -1472,6 +1517,7 @@ mod client_surface_parity_tests { nested: Option, ) -> CommandTypeField { CommandTypeField { + unsigned_integer: None, name: name.into(), type_name: type_name.into(), nullable, @@ -1885,28 +1931,28 @@ mod client_surface_parity_tests { #[cfg(feature = "sqlite")] const SQLITE_RESTRICTED_GOLDENS: ArtifactGoldens = ArtifactGoldens { - manifest: "sha256:c1c3dd3f242f82225b486542b1737976f791e019f50e015f8755f48d70685f9a", + manifest: "sha256:7c8a181165a416610142c06f6b0ddca34358d884b738af10e11bf9b928817bed", static_sdl: "sha256:03252ba251b1ddac611fe567d816f780f0876f9f6ce263be95a3480f88fc2283", runtime_sdl: "sha256:3d099b8c0b27f0dcbdd677199f767fcc5993071e06b2c0a7d4aa02caf4e5f4ac", }; #[cfg(feature = "sqlite")] const SQLITE_ADMIN_GOLDENS: ArtifactGoldens = ArtifactGoldens { - manifest: "sha256:94345b9e29bccaa77aa5083eb73014a1a102db7ce1f03f022a2f0e242b5c84d2", + manifest: "sha256:827e381234e72fa315daffa7a018f2c92f964d75ef70fbc86983b3aa704e52d9", static_sdl: "sha256:128b85bcd6485d14de62b0976e9f12e8b35ad9e7d96a5627d1edcfcabdb591b3", runtime_sdl: "sha256:c0b6d600d353357ab4f393897fb6b7ee51f69546f7a4cc4b6786e42abe621f5c", }; #[cfg(feature = "postgres")] const POSTGRES_RESTRICTED_GOLDENS: ArtifactGoldens = ArtifactGoldens { - manifest: "sha256:c1c3dd3f242f82225b486542b1737976f791e019f50e015f8755f48d70685f9a", + manifest: "sha256:7c8a181165a416610142c06f6b0ddca34358d884b738af10e11bf9b928817bed", static_sdl: "sha256:03252ba251b1ddac611fe567d816f780f0876f9f6ce263be95a3480f88fc2283", runtime_sdl: "sha256:3d099b8c0b27f0dcbdd677199f767fcc5993071e06b2c0a7d4aa02caf4e5f4ac", }; #[cfg(feature = "postgres")] const POSTGRES_ADMIN_GOLDENS: ArtifactGoldens = ArtifactGoldens { - manifest: "sha256:8ff2691f33789c8267b1338603b2ee3544841f8b17cc5b90ea2e851381dd36de", + manifest: "sha256:8e123332eb9ae3364290429e30feeee8a1da3cbad4004224382b8e6128c918a2", static_sdl: "sha256:afe92660c1700845ed5f3e0ddaacc4b481799c39f86b8eacecb46d1b8f99d421", runtime_sdl: "sha256:de4885736fdf22a57ccc55160d63b6d1a7fdb33728dec399e8a81d54ce7e8c09", }; diff --git a/src/graphql/projection_delta/runtime.rs b/src/graphql/projection_delta/runtime.rs index c8001d3bb..8ab93c6bd 100644 --- a/src/graphql/projection_delta/runtime.rs +++ b/src/graphql/projection_delta/runtime.rs @@ -810,6 +810,15 @@ fn trusted_preset_value_matches(codec: &str, value: &serde_json::Value) -> bool "json_number_precision_limited" => value .as_i64() .is_some_and(|value| (-9_007_199_254_740_991..=9_007_199_254_740_991).contains(&value)), + codec if crate::command::CommandUnsignedInteger::from_client_codec(codec).is_some() => { + crate::command::CommandUnsignedInteger::from_client_codec(codec).is_some_and( + |unsigned| { + value + .as_u64() + .is_some_and(|value| value <= unsigned.client_max_value()) + }, + ) + } "float64" => value.as_f64().is_some_and(f64::is_finite), "json" => true, _ => false, @@ -820,6 +829,27 @@ fn trusted_preset_value_matches(codec: &str, value: &serde_json::Value) -> bool mod trusted_preset_tests { use super::trusted_preset_value_matches; + #[test] + fn unsigned_presets_require_exact_bounded_integers() { + for (codec, max) in [ + ("uint8", 255_u64), + ("uint16", 65_535), + ("uint32", 4_294_967_295), + ("uint64_safe_integer", 9_007_199_254_740_991), + ] { + assert!(trusted_preset_value_matches(codec, &serde_json::json!(0))); + assert!(trusted_preset_value_matches(codec, &serde_json::json!(max))); + for invalid in [ + serde_json::json!(-1), + serde_json::json!(1.5), + serde_json::json!(max + 1), + serde_json::json!("1"), + ] { + assert!(!trusted_preset_value_matches(codec, &invalid)); + } + } + } + #[test] fn base64_preset_inventory_requires_canonical_standard_encoding() { assert!(trusted_preset_value_matches( diff --git a/src/graphql/schema.rs b/src/graphql/schema.rs index 090bbd2e2..ca5b16627 100644 --- a/src/graphql/schema.rs +++ b/src/graphql/schema.rs @@ -1221,6 +1221,7 @@ mod causal_command_schema_tests { input: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "CompleteTodoInput".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -1232,6 +1233,7 @@ mod causal_command_schema_tests { output: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "CompleteTodoPayload".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/src/graphql/sdl.rs b/src/graphql/sdl.rs index 9c7dbda27..0bcdc2903 100644 --- a/src/graphql/sdl.rs +++ b/src/graphql/sdl.rs @@ -615,6 +615,7 @@ mod causal_command_sdl_tests { input: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "CompleteTodoInput".into(), fields: vec![crate::graphql::surface::SurfaceTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -626,6 +627,7 @@ mod causal_command_sdl_tests { output: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "CompleteTodoPayload".into(), fields: vec![crate::graphql::surface::SurfaceTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/src/graphql/surface/effects.rs b/src/graphql/surface/effects.rs index a7bf2e418..8934349d0 100644 --- a/src/graphql/surface/effects.rs +++ b/src/graphql/surface/effects.rs @@ -599,6 +599,14 @@ pub(in crate::graphql::surface) fn canonicalize_type_def( definition.name, field.name )); } + if field.unsigned_integer.is_some() + && (field.type_name != "BigInt" || field.nested.is_some()) + { + return Err(format!( + "command type `{}` field `{}` has an invalid unsigned refinement", + definition.name, field.name + )); + } if let Some(nested) = &mut field.nested { canonicalize_type_def(nested)?; if field.type_name != nested.name { diff --git a/src/graphql/surface/tests.rs b/src/graphql/surface/tests.rs index 1fa560d3f..72e5fe3e8 100644 --- a/src/graphql/surface/tests.rs +++ b/src/graphql/surface/tests.rs @@ -255,6 +255,7 @@ fn test_command( input: CommandTypeDef::new( "TestCommandInput", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -289,6 +290,7 @@ fn causal_surface_commands_accept_modeled_event_selectors_but_not_empty_authorit CommandTypeDef::new( "CausalPayload", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -896,6 +898,7 @@ fn command_surface_rejects_duplicate_mutation_field_ids() { let output = CommandTypeDef::new( "TestCommandPayload", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -934,6 +937,7 @@ fn command_surface_rejects_empty_nested_and_surface_colliding_types() { CommandTypeDef::new( "OuterPayload", vec![CommandTypeField { + unsigned_integer: None, name: "inner".into(), type_name: "InnerPayload".into(), nullable: false, @@ -955,6 +959,7 @@ fn command_surface_rejects_empty_nested_and_surface_colliding_types() { CommandTypeDef::new( "OrderView", vec![CommandTypeField { + unsigned_integer: None, name: "order_id".into(), type_name: "String".into(), nullable: false, @@ -999,6 +1004,7 @@ fn projected_output_reuse_and_sdl_emission_use_the_same_exact_predicate() { let one_string_field = |name: &str| SurfaceTypeDef { name: name.into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "order_id".into(), type_name: "String".into(), nullable: false, @@ -1065,6 +1071,7 @@ fn role_surface_legacy_effects_never_become_v2_client_authority() { name: "UpdateOrderInput".into(), fields: vec![ SurfaceTypeField { + unsigned_integer: None, name: "order_id".into(), type_name: "String".into(), nullable: false, @@ -1073,6 +1080,7 @@ fn role_surface_legacy_effects_never_become_v2_client_authority() { nested: None, }, SurfaceTypeField { + unsigned_integer: None, name: "customer_id".into(), type_name: "String".into(), nullable: false, @@ -1098,6 +1106,7 @@ fn role_surface_legacy_effects_never_become_v2_client_authority() { output: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "AssignCustomerPayload".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "order_id".into(), type_name: "String".into(), nullable: false, @@ -1132,6 +1141,7 @@ fn role_surface_legacy_effects_never_become_v2_client_authority() { output: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "ApplyPresetPayload".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "order_id".into(), type_name: "String".into(), nullable: false, @@ -1938,6 +1948,7 @@ fn constant_validation_uses_exact_wire_scalar_domains() { output: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "ConstantPayload".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "ok".into(), type_name: "Boolean".into(), nullable: false, @@ -2000,6 +2011,7 @@ fn missing_surface_primary_key_column_is_a_configuration_error_not_a_panic() { output: SurfaceCommandShape::Typed(SurfaceTypeDef { name: "PatchOrderPayload".into(), fields: vec![SurfaceTypeField { + unsigned_integer: None, name: "order_id".into(), type_name: "String".into(), nullable: false, diff --git a/src/graphql/surface/types.rs b/src/graphql/surface/types.rs index 628622cae..9222fa82c 100644 --- a/src/graphql/surface/types.rs +++ b/src/graphql/surface/types.rs @@ -172,6 +172,7 @@ pub struct SurfaceTypeField { pub nullable: bool, pub list: bool, pub item_nullable: bool, + pub unsigned_integer: Option, pub nested: Option>, } @@ -1065,14 +1066,20 @@ fn command_shape_value(shape: &SurfaceCommandShape) -> serde_json::Value { fn type_def_value(definition: &SurfaceTypeDef) -> serde_json::Value { serde_json::json!({ "name": definition.name, - "fields": definition.fields.iter().map(|field| serde_json::json!({ + "fields": definition.fields.iter().map(|field| { + let mut value = serde_json::json!({ "name": field.name, "type_name": field.type_name, "nullable": field.nullable, "list": field.list, "item_nullable": field.item_nullable, "nested": field.nested.as_deref().map(type_def_value), - })).collect::>(), + }); + if let Some(unsigned) = field.unsigned_integer { + value["unsigned_integer"] = serde_json::json!(unsigned); + } + value + }).collect::>(), }) } diff --git a/src/microsvc/service/tests.rs b/src/microsvc/service/tests.rs index 0e99d74e9..6f9fcab98 100644 --- a/src/microsvc/service/tests.rs +++ b/src/microsvc/service/tests.rs @@ -82,6 +82,7 @@ fn one_string_field(name: &str, field: &str) -> CommandTypeDef { CommandTypeDef::new( name, vec![CommandTypeField { + unsigned_integer: None, name: field.into(), type_name: "String".into(), nullable: false, @@ -118,6 +119,7 @@ impl CommandInputType for CausalTestInput { "CausalTestInput", vec![ CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -126,6 +128,7 @@ impl CommandInputType for CausalTestInput { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "label".into(), type_name: "String".into(), nullable: false, @@ -614,6 +617,7 @@ impl CommandOutputType for CausalLifecycleView { "CausalLifecycleView", vec![ CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -622,6 +626,7 @@ impl CommandOutputType for CausalLifecycleView { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "label".into(), type_name: "String".into(), nullable: false, diff --git a/tests/application_composition.rs b/tests/application_composition.rs index 3fc3e970d..d19bbd421 100644 --- a/tests/application_composition.rs +++ b/tests/application_composition.rs @@ -96,6 +96,7 @@ fn command(id: &str) -> CommandSpec { CommandTypeSpec { name: format!("{id}Input"), fields: vec![CommandTypeField { + unsigned_integer: None, name: "title".into(), type_name: "String".into(), nullable: false, @@ -107,6 +108,7 @@ fn command(id: &str) -> CommandSpec { CommandTypeSpec { name: format!("{id}Output"), fields: vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/tests/application_plans.rs b/tests/application_plans.rs index 704f9f8c9..ff2c14470 100644 --- a/tests/application_plans.rs +++ b/tests/application_plans.rs @@ -14,6 +14,7 @@ fn portable_command(id: &str, consistency: CommandConsistency) -> CommandSpec { CommandTypeSpec { name: format!("{id}Input"), fields: vec![CommandTypeField { + unsigned_integer: None, name: "title".into(), type_name: "String".into(), nullable: false, @@ -25,6 +26,7 @@ fn portable_command(id: &str, consistency: CommandConsistency) -> CommandSpec { CommandTypeSpec { name: format!("{id}Output"), fields: vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/tests/causal_public_invoke/main.rs b/tests/causal_public_invoke/main.rs index 671ca8399..5d5ab054b 100644 --- a/tests/causal_public_invoke/main.rs +++ b/tests/causal_public_invoke/main.rs @@ -56,6 +56,7 @@ impl CommandInputType for CompleteInput { CommandTypeDef::new( "CompleteInput", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -78,6 +79,7 @@ impl CommandOutputType for CompletePayload { CommandTypeDef::new( "CompletePayload", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/tests/causal_wait_path/main.rs b/tests/causal_wait_path/main.rs index 9870fd4d4..ad99a05a1 100644 --- a/tests/causal_wait_path/main.rs +++ b/tests/causal_wait_path/main.rs @@ -73,6 +73,7 @@ impl CommandInputType for IdInput { CommandTypeDef::new( "IdInput", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -95,6 +96,7 @@ impl CommandOutputType for IdPayload { CommandTypeDef::new( "IdPayload", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/tests/e2e_ui_celld_nats_profile/main.rs b/tests/e2e_ui_celld_nats_profile/main.rs index d06f30526..28632b6ce 100644 --- a/tests/e2e_ui_celld_nats_profile/main.rs +++ b/tests/e2e_ui_celld_nats_profile/main.rs @@ -163,6 +163,7 @@ mod live { "CreateInput", vec![ CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -171,6 +172,7 @@ mod live { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "title".into(), type_name: "String".into(), nullable: false, @@ -194,6 +196,7 @@ mod live { CommandTypeDef::new( "IdPayload", vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/tests/fixtures/generated-draining-command-v2.json b/tests/fixtures/generated-draining-command-v2.json index 17ccf0ae8..cdd7b72ed 100644 --- a/tests/fixtures/generated-draining-command-v2.json +++ b/tests/fixtures/generated-draining-command-v2.json @@ -38,7 +38,7 @@ }, "protocol": { "operation": "sha256:bb3a777dd32f7d40cccc173fafdbc7464f5dc50a3d17587909689a880fb0205a", - "protocolHash": "sha256:00fb342f3acb4dc1c1716a43cc3001c748d5f6c500ff831690d820e9e43e2782", + "protocolHash": "sha256:0dfa8a3f49e17d8d99c5c095c1ed14f528cae3e55fbc2f2b852975a50936ec5b", "schemaHash": "sha256:7e4d542ce141b2a9f350433b03603ae0ebfecd38c15b2e9d436fe8d0bca2662b", "surface": { "kind": "role", diff --git a/tests/graphql_commands/main.rs b/tests/graphql_commands/main.rs index fac88121a..9cc534d09 100644 --- a/tests/graphql_commands/main.rs +++ b/tests/graphql_commands/main.rs @@ -10,6 +10,7 @@ fn graphql_type_def_mapping_golden() { "CreateItemInput", vec![ CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -18,6 +19,7 @@ fn graphql_type_def_mapping_golden() { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "tags".into(), type_name: "String".into(), nullable: true, @@ -48,6 +50,45 @@ struct DerivedOutput { id: String, } +#[derive(distributed::CommandInput)] +#[allow(dead_code)] +struct UnsignedInput { + tiny: u8, + short: Option, + revisions: Vec>, + revision: u64, + signed: i64, +} + +#[test] +fn unsigned_command_derive_preserves_range_and_portable_contract() { + use distributed::command::{CommandInputType, CommandUnsignedInteger as Unsigned}; + let definition = UnsignedInput::command_type(); + assert_eq!( + definition + .fields + .iter() + .map(|field| field.unsigned_integer) + .collect::>(), + vec![ + Some(Unsigned::U8), + Some(Unsigned::U16), + Some(Unsigned::U32), + Some(Unsigned::U64), + None + ] + ); + assert!(definition + .fields + .iter() + .all(|field| field.type_name == "BigInt")); + assert!(definition.fields[2].list && definition.fields[2].item_nullable); + let portable = distributed::application::CommandTypeSpec::from(&definition); + let encoded = serde_json::to_value(&portable).unwrap(); + assert_eq!(encoded["fields"][3]["unsigned_integer"], "u64"); + assert!(encoded["fields"][4].get("unsigned_integer").is_none()); +} + #[derive(distributed::CommandInput, serde::Deserialize)] #[serde(rename_all = "camelCase")] #[allow(dead_code)] diff --git a/tests/replay_event_version.rs b/tests/replay_event_version.rs new file mode 100644 index 000000000..8cf7b6452 --- /dev/null +++ b/tests/replay_event_version.rs @@ -0,0 +1,305 @@ +use distributed::{ + hydrate, hydrate_from_snapshot, Aggregate, AggregateRepository, CommitBatch, Entity, + EventRecord, RepositoryError, SnapshotRecord, SnapshotStore, Snapshottable, StreamIdentity, + StreamWrite, TransactionalCommit, +}; +use serde::{Deserialize, Serialize}; + +#[derive(Default, Serialize, Deserialize, distributed::Snapshot)] +struct Versioned { + entity: Entity, + value: String, + calls: usize, +} + +#[distributed::sourced(entity)] +impl Versioned { + #[event("renamed", version = 2)] + fn rename(&mut self, value: String) { + self.value = value; + self.calls += 1; + } + + #[event("cleared", version = 2)] + fn clear(&mut self) { + self.value.clear(); + self.calls += 1; + } +} + +#[derive(Default, Serialize, Deserialize, distributed::Snapshot)] +struct Registered { + entity: Entity, + value: String, +} + +impl Registered { + #[distributed::digest("renamed", version = 2)] + fn rename(&mut self, value: String) { + self.value = value; + } + + #[distributed::digest("cleared")] + fn clear(&mut self) { + self.value.clear(); + } +} + +distributed::aggregate!(Registered, entity { + "renamed"(value), version = 2 => rename, + "cleared"() => clear(), + "ignored"(payload), version = 2 => clear(), +}); + +#[test] +fn sourced_rejects_same_shaped_older_and_future_events_before_mutation() { + for version in [0, 1, 3, u64::MAX] { + for (name, payload) in [ + ( + "renamed", + bitcode::serialize(&("changed".to_owned(),)).unwrap(), + ), + ("cleared", vec![]), + ] { + let mut aggregate = Versioned::default(); + aggregate.rename("unchanged".into()).unwrap(); + let before = serde_json::to_value(&aggregate.entity).unwrap(); + let event = EventRecord::new_versioned(name, payload, 2, version); + assert!( + aggregate.replay_event(&event).is_err(), + "accepted {name} v{version}" + ); + assert_eq!(aggregate.value, "unchanged"); + assert_eq!(aggregate.calls, 1); + assert_eq!(serde_json::to_value(&aggregate.entity).unwrap(), before); + } + } +} + +#[test] +fn aggregate_macro_rejects_same_shaped_versions_before_mutation() { + for (name, expected) in [("renamed", 2), ("cleared", 1), ("ignored", 2)] { + for version in [expected - 1, expected + 1, u64::MAX] { + let mut aggregate = Registered::default(); + aggregate.rename("unchanged".into()).unwrap(); + let before = serde_json::to_value(&aggregate.entity).unwrap(); + let event = EventRecord::new_versioned( + name, + bitcode::serialize(&("changed".to_owned(),)).unwrap(), + 2, + version, + ); + assert!(aggregate.replay_event(&event).is_err()); + assert_eq!(aggregate.value, "unchanged"); + assert_eq!(serde_json::to_value(&aggregate.entity).unwrap(), before); + } + } +} + +fn rename_event(version: u64, sequence: u64) -> EventRecord { + EventRecord::new_versioned( + "renamed", + bitcode::serialize(&("changed".to_owned(),)).unwrap(), + sequence, + version, + ) +} + +#[test] +fn exact_version_replays_with_both_macros() { + let mut entity = Entity::new(); + entity.load_from_history(vec![rename_event(2, 1)]); + let sourced = hydrate::(entity.clone()).unwrap(); + let registered = hydrate::(entity).unwrap(); + assert_eq!(sourced.value, "changed"); + assert_eq!(sourced.calls, 1); + assert_eq!(registered.value, "changed"); + assert_eq!(sourced.entity.version(), 1); + assert!(sourced.entity.new_events().is_empty()); +} + +fn snapshot() -> SnapshotRecord { + let mut aggregate = A::new_empty(); + aggregate.entity_mut().set_id("item"); + SnapshotRecord::new( + A::aggregate_type(), + "item", + 10, + A::SNAPSHOT_VERSION, + bitcode::serialize(&aggregate.create_snapshot()).unwrap(), + ) +} + +fn check_snapshot_tail() { + for version in [1, 2, 3] { + // Production cell and native repositories load only the suffix after a + // cached snapshot; its prefix still counts toward the next write fence. + let mut entity = Entity::new(); + entity.set_id("item"); + let event = rename_event(version, 11); + entity.load_tail_from_history(vec![event.clone()], 10); + let result = hydrate_from_snapshot::(entity, snapshot::()); + if version == 2 { + let aggregate = result.unwrap(); + assert_eq!(aggregate.entity().version(), 11); + assert_eq!(aggregate.entity().events(), &[event]); + assert!(aggregate.entity().new_events().is_empty()); + } else { + assert!(matches!(result, Err(RepositoryError::Replay(message)) + if message == format!("Unsupported event version for renamed: expected 2, got {version}"))); + } + } +} + +#[test] +fn snapshot_prefix_tail_checks_versions_for_both_macros() { + check_snapshot_tail::(); + check_snapshot_tail::(); +} + +#[derive(Default, Serialize, Deserialize, distributed::Snapshot)] +struct Upcasted { + entity: Entity, + value: String, +} + +#[derive(Default)] +struct RegisteredUpcast { + entity: Entity, + value: String, +} + +impl RegisteredUpcast { + #[distributed::digest("renamed", version = 2)] + fn rename(&mut self, value: String) { + self.value = value; + } +} + +distributed::aggregate!(RegisteredUpcast, entity { + "renamed"(value), version = 2 => rename, +} upcasters [ + ("renamed", 1 => 2, (String,) => (String,), convert_semantics), +]); + +#[derive(Default)] +struct IncompleteUpcast { + entity: Entity, +} + +#[distributed::sourced(entity, upcasters( + ("renamed", 1 => 2, (String,) => (String,), convert_semantics), +))] +impl IncompleteUpcast { + #[event("renamed", version = 3)] + fn rename(&mut self, _value: String) { + self.entity.set_id("incompatible-event-applied"); + } +} + +#[test] +fn incomplete_same_shaped_upcast_chain_is_rejected() { + let mut entity = Entity::new(); + entity.load_from_history(vec![rename_event(1, 1)]); + assert!( + matches!(hydrate::(entity), Err(RepositoryError::Replay(message)) + if message == "Unsupported event version for renamed: expected 3, got 2") + ); +} + +fn convert_semantics((value,): (String,)) -> (String,) { + (format!("v2:{value}"),) +} + +#[distributed::sourced(entity, upcasters( + ("renamed", 1 => 2, (String,) => (String,), convert_semantics), +))] +impl Upcasted { + #[event("renamed", version = 2)] + fn rename(&mut self, value: String) { + self.value = value; + } +} + +#[test] +fn same_shaped_upcast_precedes_guard_and_preserves_stored_history() { + let event = rename_event(1, 1); + let mut entity = Entity::new(); + entity.load_from_history(vec![event.clone()]); + let aggregate = hydrate::(entity).unwrap(); + assert_eq!(aggregate.value, "v2:changed"); + assert_eq!(aggregate.entity.events(), &[event]); + + let mut entity = Entity::new(); + let event = rename_event(1, 1); + entity.load_from_history(vec![event.clone()]); + let aggregate = hydrate::(entity).unwrap(); + assert_eq!(aggregate.value, "v2:changed"); + assert_eq!(aggregate.entity.events(), &[event]); + + let tail = rename_event(1, 11); + let mut entity = Entity::new(); + entity.set_id("item"); + entity.load_tail_from_history(vec![tail.clone()], 10); + let aggregate = hydrate_from_snapshot::(entity, snapshot::()).unwrap(); + assert_eq!(aggregate.value, "v2:changed"); + assert_eq!(aggregate.entity.version(), 11); + assert_eq!(aggregate.entity.events(), &[tail]); +} + +#[tokio::test] +async fn cell_repository_rejects_incompatible_tail_without_changing_storage() { + use distributed::cell_host::CellStreamStore; + + for version in [1, 2, 3] { + let store = CellStreamStore::new(Versioned::aggregate_type(), "item").unwrap(); + let identity = StreamIdentity::new(Versioned::aggregate_type(), "item").unwrap(); + let mut entity = Entity::new(); + entity.set_id("item"); + for _ in 0..10 { + entity + .digest_v("renamed", 2, &("prefix".to_owned(),)) + .unwrap(); + } + entity + .digest_v("renamed", version, &("changed".to_owned(),)) + .unwrap(); + store + .commit_batch(CommitBatch::new(vec![StreamWrite::new( + identity.clone(), + &mut entity, + )])) + .await + .unwrap(); + store + .save_snapshot(&identity, snapshot::()) + .await + .unwrap(); + let before = store.durable_state().unwrap(); + let repository = AggregateRepository::<_, Versioned>::new(store.clone()).with_snapshots(10); + let result = repository.get("item").await; + if version == 2 { + let aggregate = result.unwrap().unwrap(); + assert_eq!(aggregate.value, "changed"); + assert_eq!(aggregate.calls, 1, "snapshot prefix must not replay"); + assert_eq!(aggregate.entity.version(), 11); + } else { + assert!(matches!(result, Err(RepositoryError::Replay(_)))); + } + assert_eq!(store.durable_state().unwrap(), before); + } +} + +#[test] +fn full_hydration_rejects_same_shaped_unsupported_versions() { + for version in [1, 3] { + let mut entity = Entity::new(); + entity.load_from_history(vec![EventRecord::new_versioned( + "renamed", + bitcode::serialize(&("changed".to_owned(),)).unwrap(), + 1, + version, + )]); + assert!(hydrate::(entity).is_err()); + } +} diff --git a/tests/typed_commands/main.rs b/tests/typed_commands/main.rs index 626da1b27..6e17e90b9 100644 --- a/tests/typed_commands/main.rs +++ b/tests/typed_commands/main.rs @@ -142,6 +142,7 @@ impl CommandOutputType for PlanView { "PlanView", vec![ CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, @@ -150,6 +151,7 @@ impl CommandOutputType for PlanView { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "title".into(), type_name: "String".into(), nullable: false, @@ -158,6 +160,7 @@ impl CommandOutputType for PlanView { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "count".into(), type_name: "BigInt".into(), nullable: false, @@ -166,6 +169,7 @@ impl CommandOutputType for PlanView { nested: None, }, CommandTypeField { + unsigned_integer: None, name: "status".into(), type_name: "String".into(), nullable: false, @@ -1013,6 +1017,7 @@ fn object_type(name: &str) -> CommandTypeDef { CommandTypeDef::new( name, vec![CommandTypeField { + unsigned_integer: None, name: "id".into(), type_name: "String".into(), nullable: false, diff --git a/tests/upcasting/aggregate.rs b/tests/upcasting/aggregate.rs index aaa7f0c82..e939c897b 100644 --- a/tests/upcasting/aggregate.rs +++ b/tests/upcasting/aggregate.rs @@ -71,7 +71,7 @@ impl TodoV2 { } distributed::aggregate!(TodoV2, entity, aggregate_type = "Todo" { - "initialized"(id, user_id, task, priority) => initialize, + "initialized"(id, user_id, task, priority), version = 2 => initialize, "completed"() => complete(), } upcasters [ ("initialized", 1 => 2, InitializedV1 => InitializedV2, upcast_initialized_v1_v2), @@ -120,7 +120,7 @@ impl TodoV3 { } distributed::aggregate!(TodoV3, entity, aggregate_type = "Todo" { - "initialized"(id, user_id, task, priority, due_date) => initialize, + "initialized"(id, user_id, task, priority, due_date), version = 3 => initialize, "completed"() => complete(), } upcasters [ ("initialized", 1 => 2, InitializedV1 => InitializedV2, upcast_initialized_v1_v2),