diff --git a/crates/bindings-macro/src/lib.rs b/crates/bindings-macro/src/lib.rs index 7f30fc22372..e2f7016f88e 100644 --- a/crates/bindings-macro/src/lib.rs +++ b/crates/bindings-macro/src/lib.rs @@ -159,6 +159,10 @@ mod sym { symbol!(repr); symbol!(sats); symbol!(scheduled); + symbol!(outbox); + symbol!(on_result); + symbol!(param); + symbol!(target); symbol!(unique); symbol!(update); symbol!(default); @@ -229,7 +233,10 @@ fn derive_table_helper_attr() -> Attribute { /// /// Provides helper attributes for `#[spacetimedb::table]`, so that we don't get unknown attribute errors. #[doc(hidden)] -#[proc_macro_derive(__TableHelper, attributes(sats, unique, auto_inc, primary_key, index, default))] +#[proc_macro_derive( + __TableHelper, + attributes(sats, unique, auto_inc, primary_key, target, index, default, param) +)] pub fn table_helper(input: StdTokenStream) -> StdTokenStream { schema_type(input) } diff --git a/crates/bindings-macro/src/reducer.rs b/crates/bindings-macro/src/reducer.rs index 0193f2033dd..159e5acbfaf 100644 --- a/crates/bindings-macro/src/reducer.rs +++ b/crates/bindings-macro/src/reducer.rs @@ -10,6 +10,7 @@ use syn::{FnArg, Ident, ItemFn, LitStr, PatType}; pub(crate) struct ReducerArgs { name: Option, lifecycle: Option, + on_result: Option, } enum LifecycleReducer { @@ -51,6 +52,12 @@ impl ReducerArgs { check_duplicate(&args.name, &meta)?; args.name = Some(meta.value()?.parse()?); } + sym::on_result => { + check_duplicate(&args.on_result, &meta)?; + let content; + syn::parenthesized!(content in meta.input); + args.on_result = Some(content.parse()?); + } }); Ok(()) }) @@ -101,6 +108,19 @@ pub(crate) fn reducer_impl(args: ReducerArgs, original_function: &ItemFn) -> syn assert_only_lifetime_generics(original_function, "reducers")?; let lifecycle = args.lifecycle.iter().filter_map(|lc| lc.to_lifecycle_value()); + let on_result_outbox = args.on_result.as_ref().map(|path| { + let accessor = &path.segments.last().unwrap().ident; + quote!( + const ON_RESULT_OUTBOX: Option<&'static str> = + Some(spacetimedb::rt::on_result_outbox(#path::#accessor)); + ) + }); + let on_result_typecheck = args.on_result.as_ref().map(|path| { + let accessor = &path.segments.last().unwrap().ident; + quote! { + spacetimedb::rt::on_result_typecheck(#func_name, #path::#accessor); + } + }); let typed_args = extract_typed_args(original_function)?; @@ -156,6 +176,7 @@ pub(crate) fn reducer_impl(args: ReducerArgs, original_function: &ItemFn) -> syn #(let _ = <#first_arg_ty as spacetimedb::rt::ReducerContextArg>::_ITEM;)* #(let _ = <#rest_arg_tys as spacetimedb::rt::ReducerArg>::_ITEM;)* #(let _ = <#ret_ty as spacetimedb::rt::IntoReducerResult>::into_result;)* + #on_result_typecheck } }; impl #func_name { @@ -170,6 +191,7 @@ pub(crate) fn reducer_impl(args: ReducerArgs, original_function: &ItemFn) -> syn type FnKind = spacetimedb::rt::FnKindReducer; const NAME: &'static str = #reducer_name; #(const LIFECYCLE: Option = Some(#lifecycle);)* + #on_result_outbox const ARG_NAMES: &'static [Option<&'static str>] = &[#(#opt_arg_names),*]; const INVOKE: Self::Invoke = #func_name::invoke; } diff --git a/crates/bindings-macro/src/table.rs b/crates/bindings-macro/src/table.rs index e1ff4edec8a..4e865b06bad 100644 --- a/crates/bindings-macro/src/table.rs +++ b/crates/bindings-macro/src/table.rs @@ -19,6 +19,7 @@ pub(crate) struct TableArgs { access: Option, name: Option, scheduled: Option, + outbox: Option, accessor: Ident, indices: Vec, event: Option, @@ -47,6 +48,10 @@ struct ScheduledArg { at: Option, } +struct OutboxArg { + receiver_reducer: Path, +} + struct IndexArg { accessor: Ident, canonical_name: Option, @@ -78,6 +83,7 @@ impl TableArgs { pub(crate) fn parse(input: TokenStream, struct_ident: &Ident) -> syn::Result { let mut access = None; let mut scheduled = None; + let mut outbox = None; let mut accessor = None; let mut name: Option = None; let mut indices = Vec::new(); @@ -145,6 +151,10 @@ If you're migrating from SpacetimeDB 1.*, replace `name = {sym}` with `accessor check_duplicate(&scheduled, &meta)?; scheduled = Some(ScheduledArg::parse_meta(meta)?); } + sym::outbox => { + check_duplicate(&outbox, &meta)?; + outbox = Some(OutboxArg::parse_meta(meta)?); + } sym::event => { check_duplicate(&event, &meta)?; event = Some(meta.path.span()); @@ -184,6 +194,7 @@ If you're migrating from SpacetimeDB 1.*, replace `name = {name_str_value:?}` wi Ok(TableArgs { access, scheduled, + outbox, accessor, indices, name, @@ -192,6 +203,22 @@ If you're migrating from SpacetimeDB 1.*, replace `name = {name_str_value:?}` wi } } +impl OutboxArg { + fn parse_meta(meta: ParseNestedMeta) -> syn::Result { + let mut receiver_reducer = None; + + meta.parse_nested_meta(|meta| { + check_duplicate_msg(&receiver_reducer, &meta, "can only specify one outbox receiver reducer")?; + receiver_reducer = Some(meta.path); + Ok(()) + })?; + + let receiver_reducer = receiver_reducer + .ok_or_else(|| meta.error("must specify outbox receiver reducer: outbox(receiver::reducer)"))?; + Ok(Self { receiver_reducer }) + } +} + impl ScheduledArg { fn parse_meta(meta: ParseNestedMeta) -> syn::Result { let span = meta.path.span(); @@ -563,15 +590,22 @@ impl ValidatedIndex<'_> { row_type_ident: &Ident, tbl_type_ident: &Ident, flavor: AccessorType, + backend: TokenStream, ) -> TokenStream { if self.is_unique { - self.unique_accessor(row_type_ident, tbl_type_ident, flavor) + self.unique_accessor(row_type_ident, tbl_type_ident, flavor, backend) } else { - self.non_unique_accessor(vis, row_type_ident, tbl_type_ident, flavor) + self.non_unique_accessor(vis, row_type_ident, tbl_type_ident, flavor, backend) } } - fn unique_accessor(&self, row_type_ident: &Ident, tbl_type_ident: &Ident, flavor: AccessorType) -> TokenStream { + fn unique_accessor( + &self, + row_type_ident: &Ident, + tbl_type_ident: &Ident, + flavor: AccessorType, + backend: TokenStream, + ) -> TokenStream { let col = self.kind.one_col().unwrap(); let index_ident = self.accessor_name; let vis = col.vis; @@ -589,7 +623,7 @@ impl ValidatedIndex<'_> { quote! { #[doc = #doc] #vis fn #column_ident(&self) -> #unique_ty<#tbl_token, #col_ty, __indices::#index_ident> { - #unique_ty::__new(self.__backend.clone()) + #unique_ty::__new(#backend) } } } @@ -600,6 +634,7 @@ impl ValidatedIndex<'_> { row_type_ident: &Ident, tbl_type_ident: &Ident, flavor: AccessorType, + backend: TokenStream, ) -> TokenStream { let index_ident = self.accessor_name; let cols = self.kind.columns(); @@ -630,7 +665,7 @@ impl ValidatedIndex<'_> { quote! { #[doc = #doc] #vis fn #index_ident(&self) -> #handle_ty<#tbl_token, (#(#col_tys,)*), __indices::#index_ident> { - #handle_ty::__new(self.__backend.clone()) + #handle_ty::__new(#backend) } } } @@ -740,6 +775,7 @@ struct Column<'a> { vis: &'a syn::Visibility, ident: &'a syn::Ident, ty: &'a syn::Type, + param_name: String, default_value: Option, } @@ -769,8 +805,10 @@ enum ColumnAttr { Unique(Span), AutoInc(Span), PrimaryKey(Span), + Target(Span), Index(IndexArg), Default(syn::Expr, Span), + ParamName(Ident), } impl ColumnAttr { @@ -790,6 +828,11 @@ impl ColumnAttr { } else if ident == sym::primary_key { attr.meta.require_path_only()?; Some(ColumnAttr::PrimaryKey(ident.span())) + } else if ident == sym::target { + attr.meta.require_path_only()?; + Some(ColumnAttr::Target(ident.span())) + } else if ident == sym::param { + Some(ColumnAttr::ParamName(parse_param_attr(attr)?)) } else if ident == sym::default { Some(parse_default_attr(attr, ident)?) } else { @@ -798,6 +841,20 @@ impl ColumnAttr { } } +fn parse_param_attr(attr: &syn::Attribute) -> syn::Result { + let mut name = None; + attr.parse_nested_meta(|meta| { + match_meta!(match meta { + sym::name => { + check_duplicate(&name, &meta)?; + name = Some(meta.value()?.parse()?); + } + }); + Ok(()) + })?; + name.ok_or_else(|| syn::Error::new_spanned(&attr.meta, "expected `#[param(name = parameter_name)]`")) +} + fn parse_default_attr(attr: &syn::Attribute, ident: &Ident) -> syn::Result { if let Ok(expr) = attr.parse_args::() { return Ok(ColumnAttr::Default(expr, ident.span())); @@ -869,6 +926,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R let mut unique_columns = vec![]; let mut sequenced_columns = vec![]; let mut primary_key_column = None; + let mut outbox_target_column = None; for (i, field) in fields.iter().enumerate() { let col_num = i as u16; @@ -877,7 +935,9 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R let mut unique = None; let mut auto_inc = None; let mut primary_key = None; + let mut target = None; let mut default_value = None; + let mut param_name = None; for attr in field.original_attrs { let Some(attr) = ColumnAttr::parse(attr, field_ident)? else { continue; @@ -895,11 +955,19 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R check_duplicate(&primary_key, span)?; primary_key = Some(span); } + ColumnAttr::Target(span) => { + check_duplicate(&target, span)?; + target = Some(span); + } ColumnAttr::Index(index_arg) => args.indices.push(index_arg), ColumnAttr::Default(expr, span) => { check_duplicate(&default_value, span)?; default_value = Some(expr); } + ColumnAttr::ParamName(name) => { + check_duplicate(¶m_name, name.span())?; + param_name = Some(name); + } } } @@ -917,6 +985,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R ident: field_ident, vis: field.vis, ty: field.ty, + param_name: param_name.as_ref().unwrap_or(field_ident).unraw().to_string(), default_value, }; @@ -930,6 +999,14 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R check_duplicate_msg(&primary_key_column, span, "can only have one primary key per table")?; primary_key_column = Some(column.clone()); } + if let Some(span) = target { + check_duplicate_msg( + &outbox_target_column, + span, + "can only have one target column per outbox table", + )?; + outbox_target_column = Some((column.clone(), span)); + } columns.push(column.clone()); } @@ -976,14 +1053,41 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R let tablehandle_ident = format_ident!("{}__TableHandle", table_ident); let viewhandle_ident = format_ident!("{}__ViewHandle", table_ident); + let outboxhandle_ident = format_ident!("{}__OutboxHandle", table_ident); + let public_handle_ident = if args.outbox.is_some() { + &outboxhandle_ident + } else { + &tablehandle_ident + }; let index_descs = indices.iter().map(|index| index.desc()); - let index_accessors_rw = indices - .iter() - .map(|index| index.accessor(vis, original_struct_ident, &tablehandle_ident, AccessorType::ReadWrite)); - let index_accessors_ro = indices - .iter() - .map(|index| index.accessor(vis, original_struct_ident, &tablehandle_ident, AccessorType::Read)); + let index_accessors_rw = indices.iter().map(|index| { + index.accessor( + vis, + original_struct_ident, + &tablehandle_ident, + AccessorType::ReadWrite, + quote!(self.__backend.clone()), + ) + }); + let index_accessors_ro = indices.iter().map(|index| { + index.accessor( + vis, + original_struct_ident, + &tablehandle_ident, + AccessorType::Read, + quote!(self.__backend.clone()), + ) + }); + let outbox_index_accessors = indices.iter().map(|index| { + index.accessor( + vis, + original_struct_ident, + &tablehandle_ident, + AccessorType::Read, + quote!(self.__table.__backend().clone()), + ) + }); let index_marker_types: Vec<_> = indices .iter() .map(|index| index.marker_type(vis, &tablehandle_ident, primary_key_column.as_ref())) @@ -1102,7 +1206,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R `scheduled(my_reducer, at = custom_scheduled_at)`", ) })?; - let primary_key_column = primary_key_column.ok_or_else(|| { + let primary_key_column = primary_key_column.clone().ok_or_else(|| { syn::Error::new( sched.span, "scheduled tables must have a `#[primary_key] #[auto_inc] scheduled_id: u64` column", @@ -1133,6 +1237,106 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R .unzip(); let schedule = schedule.into_iter(); + if let Some(outbox) = &args.outbox { + if let Some(scheduled) = &args.scheduled { + return Err(syn::Error::new( + scheduled.span, + "outbox tables cannot also be scheduled tables", + )); + } + let primary_key_column = primary_key_column.as_ref().ok_or_else(|| { + syn::Error::new_spanned( + &outbox.receiver_reducer, + "outbox tables must have a `#[primary_key] #[auto_inc]` column of type `u64`", + ) + })?; + if !sequenced_columns + .iter() + .any(|col| col.index == primary_key_column.index) + { + return Err(syn::Error::new_spanned( + primary_key_column.ident, + "outbox tables must have a `#[primary_key] #[auto_inc]` column of type `u64`", + )); + } + } + + let mut outbox_arg_columns = None; + let outbox_typecheck = args + .outbox + .iter() + .map(|outbox| { + let receiver_reducer = &outbox.receiver_reducer; + let Some((target_column, _)) = &outbox_target_column else { + return Err(syn::Error::new_spanned( + receiver_reducer, + "outbox tables must have a `#[target]` column", + )); + }; + let target_ty = target_column.ty; + let mut receiver_module = receiver_reducer.clone(); + receiver_module.segments.pop(); + receiver_module.segments.pop_punct(); + if receiver_module.segments.is_empty() { + return Err(syn::Error::new_spanned( + receiver_reducer, + "outbox receiver reducer must be qualified, e.g. `outbox(receiver::reducer)`", + )); + } + let mut receiver_identity = receiver_module.clone(); + receiver_identity.segments.push(parse_quote!(Identity)); + let mut receiver_outbox_typecheck = receiver_module.clone(); + let reducer_ident = receiver_reducer + .segments + .last() + .expect("receiver reducer path checked above") + .ident + .unraw(); + receiver_outbox_typecheck + .segments + .push(parse_quote!(__spacetimedb_typecheck_outbox)); + let reducer_name = receiver_outbox_typecheck + .segments + .last_mut() + .expect("receiver reducer path checked above"); + reducer_name.ident = format_ident!("__spacetimedb_typecheck_outbox_{}", reducer_ident); + let outbox_columns = columns.iter().map(|col| { + let ident = col.ident; + let ty = col.ty; + let param = Ident::new(&col.param_name, ident.span()); + let index = col.index; + quote!((#index, #ident: #ty as #param),) + }); + outbox_arg_columns = Some(quote! { + #receiver_outbox_typecheck!(#receiver_module, { #(#outbox_columns)* }) + }); + let primary_key_typecheck = primary_key_column.as_ref().map(|col| { + let ty = col.ty; + quote!(spacetimedb::rt::assert_outbox_table_primary_key::<#ty>();) + }); + Ok(quote! { + let _ = |x: #target_ty| { let _: #receiver_identity = x; }; + #primary_key_typecheck + }) + }) + .collect::>>()?; + + let outbox_desc = match (&args.outbox, &outbox_target_column) { + (Some(outbox), Some((target_column, _))) => { + let receiver_reducer = &outbox.receiver_reducer; + let target_col = target_column.index; + quote! { + const OUTBOX: Option> = Some(spacetimedb::table::OutboxDesc { + remote_reducer_name: <#receiver_reducer as spacetimedb::rt::RemoteReducer>::NAME, + arg_columns: #outbox_arg_columns, + target_column: #target_col, + signature_hash: <#receiver_reducer as spacetimedb::rt::RemoteReducer>::SIGNATURE_HASH, + }); + } + } + _ => quote! {}, + }; + let unique_err = if !unique_columns.is_empty() { quote!(spacetimedb::UniqueConstraintViolation) } else { @@ -1145,7 +1349,6 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R }; let field_types = fields.iter().map(|f| f.ty).collect::>(); - let tabletype_impl = quote! { use spacetimedb::Serialize; impl spacetimedb::Table for #tablehandle_ident { @@ -1166,8 +1369,10 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R #(const PRIMARY_KEY: Option = Some(#primary_col_id);)* const SEQUENCES: &'static [u16] = &[#(#sequence_col_ids),*]; #(const SCHEDULE: Option> = Some(#schedule);)* + #outbox_desc #table_id_from_name_func + fn __backend(&self) -> &spacetimedb::table::TableHandleBackend { &self.__backend } @@ -1187,19 +1392,30 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R } }; + let table_handle_value = quote! { + #tablehandle_ident { __backend: self.__table_backend() } + }; + let public_handle_value = if args.outbox.is_some() { + quote! { + #outboxhandle_ident { + __table: spacetimedb::table::OutboxTable::__new(#table_handle_value), + } + } + } else { + table_handle_value + }; + // Output all macro data let trait_def = quote_spanned! {table_ident.span()=> #[allow(non_camel_case_types, dead_code)] #vis trait #table_ident { #[allow(non_camel_case_types, dead_code)] - fn #table_ident(&self) -> #tablehandle_ident; + fn #table_ident(&self) -> #public_handle_ident; } impl #table_ident for spacetimedb::Local { #[allow(non_camel_case_types, dead_code)] - fn #table_ident(&self) -> #tablehandle_ident { - #tablehandle_ident { - __backend: self.__table_backend(), - } + fn #table_ident(&self) -> #public_handle_ident { + #public_handle_value } } }; @@ -1308,6 +1524,31 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R } }; + // Keep the existing Table API intact. Outboxes expose a separate facade whose + // deref target shares insertion/iteration, but never exposes the mutable handle. + let outboxhandle_def = args.outbox.is_some().then(|| { + quote! { + #[allow(non_camel_case_types)] + #[non_exhaustive] + #vis struct #outboxhandle_ident { + __table: spacetimedb::table::OutboxTable<#tablehandle_ident>, + } + } + }); + let outboxhandle_impl = args.outbox.is_some().then(|| { + quote! { + impl #outboxhandle_ident { + #(#outbox_index_accessors)* + } + impl ::core::ops::Deref for #outboxhandle_ident { + type Target = spacetimedb::table::OutboxTable<#tablehandle_ident>; + fn deref(&self) -> &Self::Target { + &self.__table + } + } + } + }); + let viewhandle_def = quote! { #[allow(non_camel_case_types)] #[non_exhaustive] @@ -1337,6 +1578,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R const _: () = { #(let _ = <#field_types as spacetimedb::rt::TableColumn>::_ITEM;)* #schedule_typecheck + #(#outbox_typecheck)* #default_type_check }; @@ -1344,11 +1586,13 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R #trait_def_view #tablehandle_def + #outboxhandle_def #viewhandle_def #table_query_handle_def #struct_level_query_impl const _: () = { + #outboxhandle_impl impl #tablehandle_ident { #(#index_accessors_rw)* } diff --git a/crates/bindings/src/rt.rs b/crates/bindings/src/rt.rs index 65e06030913..59d60c70d5f 100644 --- a/crates/bindings/src/rt.rs +++ b/crates/bindings/src/rt.rs @@ -170,6 +170,9 @@ pub trait FnInfo: ExplicitNames { /// Currently only views use this metadata. const VIEW_PRIMARY_KEY_COLUMNS: &'static [&'static str] = &[]; + /// The source name of the outbox table whose results this reducer handles. + const ON_RESULT_OUTBOX: Option<&'static str> = None; + /// The function to invoke. const INVOKE: Self::Invoke; @@ -518,6 +521,25 @@ where core::mem::forget(_x); } +/// Resolve an outbox accessor to its registered table source name. +pub const fn on_result_outbox(_table: fn(&crate::Local) -> T) -> &'static str +where + T: core::ops::Deref>, + Tbl: crate::Table, +{ + Tbl::TABLE_NAME +} + +/// Assert that an `on_result` reducer accepts the outbox row and delivery result. +pub fn on_result_typecheck(_f: F, _table: fn(&crate::Local) -> T) +where + T: core::ops::Deref>, + Tbl: crate::Table, + F: Fn(&ReducerContext, Tbl::Row, Result<(), String>) -> Ret, + Ret: IntoReducerResult, +{ +} + /// Tacit marker argument to [`ExportFunctionForScheduledTable`] for reducers. pub struct FnKindReducer { _never: Infallible, @@ -596,6 +618,9 @@ impl TableColumn for T {} /// Assert that the primary_key column of a scheduled table is a u64. pub const fn assert_scheduled_table_primary_key() {} +/// Assert that the primary_key column of an outbox table is a u64. +pub const fn assert_outbox_table_primary_key() {} + mod sealed { pub trait Sealed {} } @@ -607,6 +632,13 @@ pub trait ScheduledTablePrimaryKey: sealed::Sealed {} impl sealed::Sealed for u64 {} impl ScheduledTablePrimaryKey for u64 {} +#[diagnostic::on_unimplemented( + message = "outbox table primary key must be a `u64`", + label = "should be `u64`, not `{Self}`" +)] +pub trait OutboxTablePrimaryKey: sealed::Sealed {} +impl OutboxTablePrimaryKey for u64 {} + /// Used in the last type parameter of `Reducer` to indicate that the /// context argument *should* be passed to the reducer logic. pub struct ContextArg; @@ -825,6 +857,18 @@ pub fn register_table() { table.finish(); + if let Some(outbox) = T::OUTBOX { + module.inner.add_outbox( + T::TABLE_NAME, + outbox.remote_reducer_name, + outbox.target_column, + outbox.arg_columns.iter().copied().map(Into::into), + None::<&str>, + spacetimedb_lib::Hash::from_hex(outbox.signature_hash) + .expect("generated outbox signature hash must be valid hex"), + ); + } + module.inner.add_explicit_names(T::explicit_names()); }) } @@ -852,6 +896,9 @@ pub fn register_reducer<'a, A: Args<'a>, I: FnInfo>(_: impl } else { module.inner.add_reducer(I::NAME, params); } + if let Some(outbox_table) = I::ON_RESULT_OUTBOX { + module.on_result_reducers.push((outbox_table, I::NAME)); + } module.reducers.push(I::INVOKE); module.inner.add_explicit_names(I::explicit_names()); @@ -1133,6 +1180,7 @@ pub struct ModuleBuilder { views: Vec, /// The anonymous views of the module. views_anon: Vec, + on_result_reducers: Vec<(&'static str, &'static str)>, } // Not actually a mutex; because WASM is single-threaded this basically just turns into a refcell. @@ -1162,11 +1210,20 @@ static ANONYMOUS_VIEWS: OnceLock> = OnceLock::new(); #[cfg(feature = "test-utils")] #[doc(hidden)] pub fn module_def_for_tests() -> RawModuleDef { + RawModuleDef::V10(describe_module().inner.finish()) +} + +fn describe_module() -> ModuleBuilder { let mut module = ModuleBuilder::default(); for describer in &mut *DESCRIBERS.lock().unwrap() { describer(&mut module) } - RawModuleDef::V10(module.inner.finish()) + + // Tables and reducers can register in either order, so link callbacks last. + for (outbox_table, reducer) in module.on_result_reducers.drain(..) { + module.inner.set_outbox_on_result(outbox_table, reducer); + } + module } /// Called by the host when the module is initialized @@ -1186,11 +1243,7 @@ pub fn module_def_for_tests() -> RawModuleDef { /// After initialization, the module cannot alter the schema. #[unsafe(no_mangle)] extern "C" fn __describe_module__(description: BytesSink) { - // Collect the `module`. - let mut module = ModuleBuilder::default(); - for describer in &mut *DESCRIBERS.lock().unwrap() { - describer(&mut module) - } + let module = describe_module(); // Serialize the module to bsatn. let module_def = module.inner.finish(); diff --git a/crates/bindings/src/table.rs b/crates/bindings/src/table.rs index aaa678b40db..8d7640c139b 100644 --- a/crates/bindings/src/table.rs +++ b/crates/bindings/src/table.rs @@ -319,8 +319,8 @@ impl TableHandleBackend { } /// Implemented for every `TableHandle` struct generated by the [`table`](macro@crate::table) macro. -/// Contains methods that are present for every table, regardless of what unique constraints -/// and indexes are present. +/// Provides common table operations, regardless of what unique constraints and indexes +/// are present. /// /// To get a `TableHandle` // TODO: should we rename this `TableHandle`? Documenting this, I think that's much clearer. @@ -463,6 +463,48 @@ pub trait Table: TableInternal + ExplicitNames { fn integrate_generated_columns(row: &mut Self::Row, generated_cols: &[u8]); } +/// An insert-only table handle used by outboxes. +/// +/// Shares ordinary table insertion and iteration without exposing deletion or updates. +/// Outbox index accessors likewise return read-only index handles. +pub struct OutboxTable { + table: T, +} + +impl OutboxTable { + #[doc(hidden)] + pub fn __new(table: T) -> Self { + Self { table } + } + + #[doc(hidden)] + pub fn __backend(&self) -> &TableHandleBackend { + self.table.__backend() + } + + /// Returns the number of pending rows. + pub fn count(&self) -> u64 { + self.table.count() + } + + /// Iterates over pending rows. + pub fn iter(&self) -> impl Iterator { + self.table.iter() + } + + /// Inserts a row, returning it with generated column values. + #[track_caller] + pub fn insert(&self, row: T::Row) -> T::Row { + self.table.insert(row) + } + + /// Inserts a row, returning any constraint violation. + #[track_caller] + pub fn try_insert(&self, row: T::Row) -> Result> { + self.table.try_insert(row) + } +} + #[doc(hidden)] #[inline] pub fn count() -> u64 { @@ -478,6 +520,7 @@ pub trait TableInternal: Sized { const PRIMARY_KEY: Option = None; const SEQUENCES: &'static [u16]; const SCHEDULE: Option> = None; + const OUTBOX: Option> = None; const IS_EVENT: bool = false; /// Returns the ID of this table. @@ -504,6 +547,14 @@ pub enum IndexAlgo<'a> { Direct { column: u16 }, } +pub struct OutboxDesc<'a> { + pub remote_reducer_name: &'a str, + /// Outbox column positions in remote reducer parameter order. + pub arg_columns: &'a [u16], + pub target_column: u16, + pub signature_hash: &'a str, +} + pub struct ScheduleDesc<'a> { pub reducer_or_procedure_name: &'a str, pub scheduled_at_column: u16, diff --git a/crates/bindings/tests/fixtures/idc_receiver/lib.rs b/crates/bindings/tests/fixtures/idc_receiver/lib.rs new file mode 100644 index 00000000000..d12dfc20386 --- /dev/null +++ b/crates/bindings/tests/fixtures/idc_receiver/lib.rs @@ -0,0 +1,127 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] + +use spacetimedb::spacetimedb_lib as __lib; + +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)] +pub struct Identity(pub spacetimedb::Identity); + +impl spacetimedb::SpacetimeType for Identity { + fn make_type( + typespace: &mut S, + ) -> spacetimedb::spacetimedb_lib::AlgebraicType { + ::make_type(typespace) + } +} + +impl spacetimedb::Serialize for Identity { + fn serialize( + &self, + serializer: S, + ) -> ::core::result::Result { + ::serialize(&self.0, serializer) + } +} + +impl<'de> spacetimedb::Deserialize<'de> for Identity { + fn deserialize>( + deserializer: D, + ) -> ::core::result::Result { + ::deserialize(deserializer).map(Self) + } +} + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct ReceivePingArgs { + pub body: ::std::string::String, +} + +#[doc(hidden)] +macro_rules! __spacetimedb_typecheck_outbox_receive_ping_body_reject_duplicate { + ($module:ident;) => {}; + ($module:ident; ($index:literal, $field:ident : $ty:ty as body), $($rest:tt)*) => { + compile_error!(concat!( + "outbox has multiple columns matching remote reducer parameter `", + stringify!(body), + "`" + )); + }; + ($module:ident; ($index:literal, $field:ident : $ty:ty as $other:ident), $($rest:tt)*) => { + $module::__spacetimedb_typecheck_outbox_receive_ping_body_reject_duplicate!($module; $($rest)*); + }; +} + +#[doc(hidden)] +pub(crate) use __spacetimedb_typecheck_outbox_receive_ping_body_reject_duplicate; + +#[doc(hidden)] +macro_rules! __spacetimedb_typecheck_outbox_receive_ping_body { + ($module:ident;) => {{ + compile_error!(concat!( + "outbox has no column matching remote reducer parameter `", + stringify!(body), + "`" + )); + 0u16 + }}; + ($module:ident; ($index:literal, $field:ident : $ty:ty as body), $($rest:tt)*) => {{ + let _ = |x: $ty| { let _: ::std::string::String = x; }; + $module::__spacetimedb_typecheck_outbox_receive_ping_body_reject_duplicate!($module; $($rest)*); + $index + }}; + ($module:ident; ($index:literal, $field:ident : $ty:ty as $other:ident), $($rest:tt)*) => { + $module::__spacetimedb_typecheck_outbox_receive_ping_body!($module; $($rest)*) + }; +} + +#[doc(hidden)] +pub(crate) use __spacetimedb_typecheck_outbox_receive_ping_body; + +#[doc(hidden)] +macro_rules! __spacetimedb_typecheck_outbox_receive_ping { + ($module:path, { $($columns:tt)* }) => {{ + #[allow(unused_imports)] + use $module as __outbox_receiver; + &[__outbox_receiver::__spacetimedb_typecheck_outbox_receive_ping_body!(__outbox_receiver; $($columns)*)] + }}; +} + +#[doc(hidden)] +pub(crate) use __spacetimedb_typecheck_outbox_receive_ping; + +#[allow(non_camel_case_types)] +pub struct receive_ping; + +impl spacetimedb::rt::RemoteReducer for receive_ping { + const NAME: &'static str = "wire_ping"; + const SIGNATURE_HASH: &'static str = "386957a94632958d5ae8343e258f3a6f9dce9d7dafc46831050894baa7a4e4e1"; + type Args = ReceivePingArgs; +} + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct NoopArgs {} + +#[doc(hidden)] +macro_rules! __spacetimedb_typecheck_outbox_noop { + ($module:path, { $($columns:tt)* }) => {{ + #[allow(unused_imports)] + use $module as __outbox_receiver; + &[] + }}; +} + +#[doc(hidden)] +pub(crate) use __spacetimedb_typecheck_outbox_noop; + +#[allow(non_camel_case_types)] +pub struct noop; + +impl spacetimedb::rt::RemoteReducer for noop { + const NAME: &'static str = "noop"; + const SIGNATURE_HASH: &'static str = "bc436888fd947d2bfd49ed6bddad31f0245c34fd42c85f9360fa0b477aa957fc"; + type Args = NoopArgs; +} diff --git a/crates/bindings/tests/pass/outbox_arg_columns.rs b/crates/bindings/tests/pass/outbox_arg_columns.rs new file mode 100644 index 00000000000..215dc7eeb5f --- /dev/null +++ b/crates/bindings/tests/pass/outbox_arg_columns.rs @@ -0,0 +1,142 @@ +#[path = "../fixtures/idc_receiver/lib.rs"] +mod receiver; + +mod remote_bindings { + pub(crate) use crate::receiver; +} + +#[spacetimedb::table( + accessor = outbound_ping, + name = "WireOutbox", + outbox(crate::remote_bindings::receiver::receive_ping), + index(accessor = by_audit, btree(columns = [audit])), + index(accessor = by_audit_hash, hash(columns = [audit])), +)] +pub struct OutboundPing { + #[primary_key] + #[auto_inc] + id: u64, + audit: String, + #[param(name = body)] + payload: String, + #[target] + destination: receiver::Identity, +} + +mod outboxes { + use super::receiver; + + #[spacetimedb::table(accessor = plain_ping, outbox(self::receiver::receive_ping))] + pub struct PlainPing { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, + body: String, + } +} + +#[spacetimedb::table(accessor = empty_ping, outbox(receiver::noop))] +pub struct EmptyPing { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, +} + +#[spacetimedb::reducer(on_result(self::outbound_ping))] +pub fn on_ping_result(_: &spacetimedb::ReducerContext, _: OutboundPing, _: Result<(), String>) {} + +mod callbacks { + #[spacetimedb::reducer(name = "WireCallback", on_result(super::outboxes::plain_ping))] + pub fn on_plain_result(_: &spacetimedb::ReducerContext, _: super::outboxes::PlainPing, _: Result<(), String>) {} +} + +fn main() { + use spacetimedb::rt::FnInfo; + use spacetimedb::table::TableInternal; + + assert_eq!(on_ping_result::ON_RESULT_OUTBOX, Some("outbound_ping")); + assert_eq!(callbacks::on_plain_result::ON_RESULT_OUTBOX, Some("plain_ping")); + assert_eq!(outbound_ping__TableHandle::OUTBOX.unwrap().arg_columns, &[2]); + assert_eq!(outbound_ping__TableHandle::OUTBOX.unwrap().target_column, 3); + assert_eq!(outboxes::plain_ping__TableHandle::OUTBOX.unwrap().arg_columns, &[2]); + assert!(empty_ping__TableHandle::OUTBOX.unwrap().arg_columns.is_empty()); + assert_eq!( + outboxes::plain_ping__TableHandle::OUTBOX.unwrap().remote_reducer_name, + "wire_ping" + ); +} + +// Outbox tables retain inserts and reads, including primary-key lookups. +fn read_and_insert(ctx: &spacetimedb::ReducerContext, row: OutboundPing) { + let table = ctx.db.outbound_ping(); + let row = table.insert(row); + let _ = table.id().find(row.id); + let _ = table.by_audit().filter("audit"); + let _ = table.by_audit_hash().filter("audit"); + let _ = table.try_insert(row); + let _ = table.count(); + let _ = table.iter(); +} + +// Existing Table bounds, associated-type paths, and UFCS calls remain sufficient. +fn mutate_generic(table: &T, row: ::Row) { + let row = spacetimedb::Table::insert(table, row); + table.delete(row); + table.clear(); +} + +fn mutate_generic_index(column: &spacetimedb::UniqueColumn, row: T::Row) +where + T: spacetimedb::Table, + C: spacetimedb::table::Index + spacetimedb::table::Column + spacetimedb::table::PrimaryKey, +{ + let row = column.update(row); + #[cfg(feature = "unstable")] + let row = column.insert_or_update(row); + #[cfg(feature = "unstable")] + let row = column.try_insert_or_update(row).unwrap(); + column.delete(C::get_field(&row)); +} + +fn mutate_generic_indexes( + ranged: &spacetimedb::RangedIndex, + pointed: &spacetimedb::PointIndex, +) where + T: spacetimedb::Table, + R: spacetimedb::table::IndexIsRanged, + P: spacetimedb::table::IndexIsPointed, +{ + ranged.delete("value"); + pointed.delete("value"); +} + +#[spacetimedb::table( + accessor = ordinary, + index(accessor = by_value, btree(columns = [value])), + index(accessor = by_value_hash, hash(columns = [value])), +)] +pub struct Ordinary { + #[primary_key] + id: u64, + value: String, +} + +// Ordinary tables retain all update and delete APIs. +fn mutate_ordinary(ctx: &spacetimedb::ReducerContext, row: Ordinary) { + use spacetimedb::Table; + let table = ctx.db.ordinary(); + let row = table.id().update(row); + #[cfg(feature = "unstable")] + let row = table.id().insert_or_update(row); + #[cfg(feature = "unstable")] + let row = table.id().try_insert_or_update(row).unwrap(); + table.delete(row); + table.id().delete(1); + table.by_value().delete("value"); + table.by_value_hash().delete("value"); + table.clear(); +} diff --git a/crates/bindings/tests/test_utils.rs b/crates/bindings/tests/test_utils.rs index 33a6edfd237..c8ac1848cb7 100644 --- a/crates/bindings/tests/test_utils.rs +++ b/crates/bindings/tests/test_utils.rs @@ -2,6 +2,9 @@ use spacetimedb::spacetimedb_lib::RawModuleDef; use spacetimedb::test_utils::{TestAuth, TestQueryError}; use spacetimedb::{reducer, table, AnonymousViewContext, Query, ReducerContext, Table, Timestamp, ViewContext}; +#[path = "fixtures/idc_receiver/lib.rs"] +mod receiver; + #[table(accessor = test_utils_user, public)] #[derive(Debug, PartialEq, Eq)] pub struct TestUtilsUser { @@ -24,6 +27,21 @@ pub fn add_test_utils_user(_ctx: &ReducerContext, id: u64, name: String) { let _ = (id, name); } +#[table(accessor = test_utils_outbox, outbox(receiver::receive_ping))] +pub struct TestUtilsOutbox { + #[primary_key] + #[auto_inc] + id: u64, + audit: String, + #[param(name = body)] + payload: String, + #[target] + destination: receiver::Identity, +} + +#[reducer(on_result(test_utils_outbox))] +pub fn on_test_utils_outbox_result(_ctx: &ReducerContext, _row: TestUtilsOutbox, _result: Result<(), String>) {} + fn query_test_utils_users_by_name(from: spacetimedb::QueryBuilder, name: &str) -> impl Query { from.test_utils_user() .r#where(|user| user.name.eq(name.to_owned())) @@ -45,6 +63,7 @@ fn module_def_includes_native_test_registrations() { let mut table_names = spacetimedb::test_utils::all_table_names(); table_names.sort_unstable(); assert!(table_names.contains(&"test_utils_user")); + assert!(table_names.contains(&"test_utils_outbox")); let RawModuleDef::V10(module) = spacetimedb::test_utils::module_def() else { panic!("test utils should return a v10 raw module def"); @@ -57,11 +76,33 @@ fn module_def_includes_native_test_registrations() { assert!(tables .iter() .any(|table| table.source_name.as_ref() == "test_utils_event")); + assert!(tables + .iter() + .any(|table| table.source_name.as_ref() == "test_utils_outbox")); let reducers = module.reducers().expect("reducers section should be present"); assert!(reducers .iter() .any(|reducer| reducer.source_name.as_ref() == "add_test_utils_user")); + assert!(reducers + .iter() + .any(|reducer| reducer.source_name.as_ref() == "on_test_utils_outbox_result")); + + let outboxes = module.outboxes().expect("outboxes section should be present"); + assert_eq!(outboxes.len(), 1); + let outbox = &outboxes[0]; + assert_eq!(outbox.table_name.as_ref(), "test_utils_outbox"); + assert_eq!(outbox.remote_reducer.as_ref(), "wire_ping"); + assert_eq!(outbox.target_column, 3.into()); + assert_eq!(outbox.arg_columns, vec![2.into()]); + assert_eq!( + outbox.signature_hash, + spacetimedb::spacetimedb_lib::Hash::from_hex( + ::SIGNATURE_HASH + ) + .unwrap() + ); + assert_eq!(outbox.on_result_reducer.as_deref(), Some("on_test_utils_outbox_result")); } #[test] diff --git a/crates/bindings/tests/ui/outbox_mutation.rs b/crates/bindings/tests/ui/outbox_mutation.rs new file mode 100644 index 00000000000..3a2cc5c3447 --- /dev/null +++ b/crates/bindings/tests/ui/outbox_mutation.rs @@ -0,0 +1,42 @@ +use spacetimedb::Table; + +#[path = "../fixtures/idc_receiver/lib.rs"] +mod receiver; + +#[spacetimedb::table( + accessor = outbound_ping, + outbox(receiver::receive_ping), + index(accessor = by_body, btree(columns = [body])), + index(accessor = by_body_hash, hash(columns = [body])), +)] +pub struct OutboundPing { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, + body: String, +} + +fn mutate(ctx: &spacetimedb::ReducerContext, row: OutboundPing) { + let table = ctx.db.outbound_ping(); + table.clear(); + table.id().delete(1); + table.by_body().delete("body"); + table.by_body_hash().delete("body"); + table.id().update(row); +} + +fn delete(ctx: &spacetimedb::ReducerContext, row: OutboundPing) { + ctx.db.outbound_ping().delete(row); +} + +fn upsert(ctx: &spacetimedb::ReducerContext, row: OutboundPing) { + ctx.db.outbound_ping().id().insert_or_update(row); +} + +fn try_upsert(ctx: &spacetimedb::ReducerContext, row: OutboundPing) { + let _ = ctx.db.outbound_ping().id().try_insert_or_update(row); +} + +fn main() {} diff --git a/crates/bindings/tests/ui/outbox_mutation.stderr b/crates/bindings/tests/ui/outbox_mutation.stderr new file mode 100644 index 00000000000..3c0c3dce702 --- /dev/null +++ b/crates/bindings/tests/ui/outbox_mutation.stderr @@ -0,0 +1,65 @@ +error[E0599]: no method named `clear` found for struct `outbound_ping__OutboxHandle` in the current scope + --> tests/ui/outbox_mutation.rs:23:11 + | + 6 | / #[spacetimedb::table( + 7 | | accessor = outbound_ping, + | |____________________________- method `clear` not found for this struct +... +23 | table.clear(); + | ^^^^^ method not found in `outbound_ping__OutboxHandle` + | + = help: items from traits can only be used if the trait is implemented and in scope + = note: the following traits define an item `clear`, perhaps you need to implement one of them: + candidate #1: `bitflags::traits::Flags` + candidate #2: `protobuf::message::Message` + candidate #3: `spacetimedb::Table` + +error[E0599]: no method named `delete` found for struct `UniqueColumnReadOnly` in the current scope + --> tests/ui/outbox_mutation.rs:24:16 + | +24 | table.id().delete(1); + | ^^^^^^ method not found in `UniqueColumnReadOnly` + +error[E0599]: no method named `delete` found for struct `RangedIndexReadOnly` in the current scope + --> tests/ui/outbox_mutation.rs:25:21 + | +25 | table.by_body().delete("body"); + | ^^^^^^ method not found in `RangedIndexReadOnly` + +error[E0599]: no method named `delete` found for struct `PointIndexReadOnly` in the current scope + --> tests/ui/outbox_mutation.rs:26:26 + | +26 | table.by_body_hash().delete("body"); + | ^^^^^^ method not found in `PointIndexReadOnly` + +error[E0599]: no method named `update` found for struct `UniqueColumnReadOnly` in the current scope + --> tests/ui/outbox_mutation.rs:27:16 + | +27 | table.id().update(row); + | ^^^^^^ method not found in `UniqueColumnReadOnly` + +error[E0599]: no method named `delete` found for struct `outbound_ping__OutboxHandle` in the current scope + --> tests/ui/outbox_mutation.rs:31:28 + | + 6 | / #[spacetimedb::table( + 7 | | accessor = outbound_ping, + | |____________________________- method `delete` not found for this struct +... +31 | ctx.db.outbound_ping().delete(row); + | ^^^^^^ method not found in `outbound_ping__OutboxHandle` + | + = help: items from traits can only be used if the trait is implemented and in scope + = note: the following trait defines an item `delete`, perhaps you need to implement it: + candidate #1: `spacetimedb::Table` + +error[E0599]: no method named `insert_or_update` found for struct `UniqueColumnReadOnly` in the current scope + --> tests/ui/outbox_mutation.rs:35:33 + | +35 | ctx.db.outbound_ping().id().insert_or_update(row); + | ^^^^^^^^^^^^^^^^ method not found in `UniqueColumnReadOnly` + +error[E0599]: no method named `try_insert_or_update` found for struct `UniqueColumnReadOnly` in the current scope + --> tests/ui/outbox_mutation.rs:39:41 + | +39 | let _ = ctx.db.outbound_ping().id().try_insert_or_update(row); + | ^^^^^^^^^^^^^^^^^^^^ method not found in `UniqueColumnReadOnly` diff --git a/crates/bindings/tests/ui/outbox_receiver_typecheck.rs b/crates/bindings/tests/ui/outbox_receiver_typecheck.rs new file mode 100644 index 00000000000..55cb1e0d732 --- /dev/null +++ b/crates/bindings/tests/ui/outbox_receiver_typecheck.rs @@ -0,0 +1,56 @@ +#[path = "../fixtures/idc_receiver/lib.rs"] +mod receiver; +#[spacetimedb::table(accessor = outbound_ping, outbox(receiver::receive_ping))] +pub struct OutboundPing { + #[primary_key] + #[auto_inc] + id: u64, + body: String, + #[target] + target: u8, +} + +#[spacetimedb::table(accessor = wrong_body, outbox(receiver::receive_ping))] +pub struct WrongBody { + #[primary_key] + #[auto_inc] + id: u64, + body: u64, + #[target] + target: receiver::Identity, +} + +#[spacetimedb::table(accessor = missing_body, outbox(receiver::receive_ping))] +pub struct MissingBody { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, + payload: String, +} + +#[spacetimedb::table(accessor = duplicate_body, outbox(receiver::receive_ping))] +pub struct DuplicateBody { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, + body: String, + #[param(name = body)] + payload: String, +} + +#[spacetimedb::table(accessor = canonical_param_name, outbox(receiver::receive_ping))] +pub struct CanonicalParamName { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, + #[param(name = Body)] + payload: String, +} + +fn main() {} diff --git a/crates/bindings/tests/ui/outbox_receiver_typecheck.stderr b/crates/bindings/tests/ui/outbox_receiver_typecheck.stderr new file mode 100644 index 00000000000..5bc3baf3a49 --- /dev/null +++ b/crates/bindings/tests/ui/outbox_receiver_typecheck.stderr @@ -0,0 +1,77 @@ +error: outbox has no column matching remote reducer parameter `body` + --> tests/ui/../fixtures/idc_receiver/lib.rs + | + | / compile_error!(concat!( + | | "outbox has no column matching remote reducer parameter `", + | | stringify!(body), + | | "`" + | | )); + | |__________^ + | + ::: tests/ui/outbox_receiver_typecheck.rs:45:1 + | +45 | #[spacetimedb::table(accessor = canonical_param_name, outbox(receiver::receive_ping))] + | -------------------------------------------------------------------------------------- in this attribute macro expansion + | + = note: this error originates in the macro `__outbox_receiver::__spacetimedb_typecheck_outbox_receive_ping_body` which comes from the expansion of the attribute macro `spacetimedb::table` (in Nightly builds, run with -Z macro-backtrace for more info) + +error: outbox has multiple columns matching remote reducer parameter `body` + --> tests/ui/../fixtures/idc_receiver/lib.rs + | + | / compile_error!(concat!( + | | "outbox has multiple columns matching remote reducer parameter `", + | | stringify!(body), + | | "`" + | | )); + | |__________^ + | + ::: tests/ui/outbox_receiver_typecheck.rs:33:1 + | +33 | #[spacetimedb::table(accessor = duplicate_body, outbox(receiver::receive_ping))] + | -------------------------------------------------------------------------------- in this attribute macro expansion + | + = note: this error originates in the macro `__outbox_receiver::__spacetimedb_typecheck_outbox_receive_ping_body_reject_duplicate` which comes from the expansion of the attribute macro `spacetimedb::table` (in Nightly builds, run with -Z macro-backtrace for more info) + +error: outbox has no column matching remote reducer parameter `body` + --> tests/ui/../fixtures/idc_receiver/lib.rs + | + | / compile_error!(concat!( + | | "outbox has no column matching remote reducer parameter `", + | | stringify!(body), + | | "`" + | | )); + | |__________^ + | + ::: tests/ui/outbox_receiver_typecheck.rs:23:1 + | +23 | #[spacetimedb::table(accessor = missing_body, outbox(receiver::receive_ping))] + | ------------------------------------------------------------------------------ in this attribute macro expansion + | + = note: this error originates in the macro `__outbox_receiver::__spacetimedb_typecheck_outbox_receive_ping_body` which comes from the expansion of the attribute macro `spacetimedb::table` (in Nightly builds, run with -Z macro-backtrace for more info) + +error[E0308]: mismatched types + --> tests/ui/outbox_receiver_typecheck.rs:3:1 + | +3 | #[spacetimedb::table(accessor = outbound_ping, outbox(receiver::receive_ping))] + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected `Identity`, found `u8` + | + = note: this error originates in the attribute macro `spacetimedb::table` (in Nightly builds, run with -Z macro-backtrace for more info) + +error[E0308]: mismatched types + --> tests/ui/../fixtures/idc_receiver/lib.rs + | + | let _ = |x: $ty| { let _: ::std::string::String = x; }; + | --------------------- ^ expected `String`, found `u64` + | | + | expected due to this + | + ::: tests/ui/outbox_receiver_typecheck.rs:13:1 + | +13 | #[spacetimedb::table(accessor = wrong_body, outbox(receiver::receive_ping))] + | ---------------------------------------------------------------------------- in this attribute macro expansion + | + = note: this error originates in the macro `__outbox_receiver::__spacetimedb_typecheck_outbox_receive_ping_body` which comes from the expansion of the attribute macro `spacetimedb::table` (in Nightly builds, run with -Z macro-backtrace for more info) +help: try using a conversion method + | +71 | let _ = |x: $ty| { let _: ::std::string::String = x.to_string(); }; + | ++++++++++++ diff --git a/crates/bindings/tests/ui/outbox_table_shape.rs b/crates/bindings/tests/ui/outbox_table_shape.rs new file mode 100644 index 00000000000..7a79dd71aec --- /dev/null +++ b/crates/bindings/tests/ui/outbox_table_shape.rs @@ -0,0 +1,53 @@ +#[path = "../fixtures/idc_receiver/lib.rs"] +mod receiver; +#[spacetimedb::table(accessor = missing_pk, outbox(receiver::receive_ping))] +pub struct MissingPk { + #[target] + target: receiver::Identity, + body: String, +} + +#[spacetimedb::table(accessor = non_auto_inc_pk, outbox(receiver::receive_ping))] +pub struct NonAutoIncPk { + #[primary_key] + id: u64, + #[target] + target: receiver::Identity, + body: String, +} + +#[spacetimedb::table(accessor = wrong_pk_type, outbox(receiver::receive_ping))] +pub struct WrongPkType { + #[primary_key] + #[auto_inc] + id: u32, + #[target] + target: receiver::Identity, + body: String, +} + +#[spacetimedb::table(accessor = scheduled_and_outbox, scheduled(scheduled), outbox(receiver::receive_ping))] +pub struct ScheduledAndOutbox { + #[primary_key] + #[auto_inc] + id: u64, + scheduled_at: spacetimedb::ScheduleAt, + #[target] + target: receiver::Identity, + body: String, +} + +#[spacetimedb::table(accessor = callback_outbox, outbox(receiver::receive_ping))] +pub struct CallbackOutbox { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: receiver::Identity, + body: String, +} + +#[spacetimedb::reducer(on_result(callback_outbox))] +pub fn bad_on_result(_ctx: &spacetimedb::ReducerContext, _row: CallbackOutbox) {} + +fn main() {} diff --git a/crates/bindings/tests/ui/outbox_table_shape.stderr b/crates/bindings/tests/ui/outbox_table_shape.stderr new file mode 100644 index 00000000000..7792aee57a1 --- /dev/null +++ b/crates/bindings/tests/ui/outbox_table_shape.stderr @@ -0,0 +1,55 @@ +error: outbox tables must have a `#[primary_key] #[auto_inc]` column of type `u64` + --> tests/ui/outbox_table_shape.rs:3:52 + | +3 | #[spacetimedb::table(accessor = missing_pk, outbox(receiver::receive_ping))] + | ^^^^^^^^^^^^^^^^^^^^^^ + +error: outbox tables must have a `#[primary_key] #[auto_inc]` column of type `u64` + --> tests/ui/outbox_table_shape.rs:13:5 + | +13 | id: u64, + | ^^ + +error: outbox tables cannot also be scheduled tables + --> tests/ui/outbox_table_shape.rs:29:55 + | +29 | #[spacetimedb::table(accessor = scheduled_and_outbox, scheduled(scheduled), outbox(receiver::receive_ping))] + | ^^^^^^^^^ + +error[E0277]: outbox table primary key must be a `u64` + --> tests/ui/outbox_table_shape.rs:23:9 + | +23 | id: u32, + | ^^^ should be `u64`, not `u32` + | + = help: the trait `spacetimedb::rt::OutboxTablePrimaryKey` is not implemented for `u32` +help: the trait `spacetimedb::rt::OutboxTablePrimaryKey` is implemented for `u64` + --> src/rt.rs + | + | impl OutboxTablePrimaryKey for u64 {} + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +note: required by a bound in `spacetimedb::rt::assert_outbox_table_primary_key` + --> src/rt.rs + | + | pub const fn assert_outbox_table_primary_key() {} + | ^^^^^^^^^^^^^^^^^^^^^ required by this bound in `assert_outbox_table_primary_key` + +error[E0593]: function is expected to take 3 arguments, but it takes 2 arguments + --> tests/ui/outbox_table_shape.rs:51:8 + | +50 | #[spacetimedb::reducer(on_result(callback_outbox))] + | --------------------------------------------------- required by a bound introduced by this call +51 | pub fn bad_on_result(_ctx: &spacetimedb::ReducerContext, _row: CallbackOutbox) {} + | -------^^^^^^^^^^^^^---------------------------------------------------------- + | | | + | | expected function that takes 3 arguments + | takes 2 arguments + | +note: required by a bound in `spacetimedb::rt::on_result_typecheck` + --> src/rt.rs + | + | pub fn on_result_typecheck(_f: F, _table: fn(&crate::Local) -> T) + | ------------------- required by a bound in this function +... + | F: Fn(&ReducerContext, Tbl::Row, Result<(), String>) -> Ret, + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `on_result_typecheck` diff --git a/crates/cli/src/subcommands/publish.rs b/crates/cli/src/subcommands/publish.rs index b7311745d89..a88919def4a 100644 --- a/crates/cli/src/subcommands/publish.rs +++ b/crates/cli/src/subcommands/publish.rs @@ -95,6 +95,7 @@ pub fn build_publish_schema(command: &clap::Command) -> Result) { + for body in bodies { + enqueue_ping(ctx, target, body); + } +} + +#[spacetimedb::table(accessor = ping_result)] +pub struct PingResult { + #[primary_key] + id: u64, + #[auto_inc] + callback_order: u64, + module_version: u32, + body: String, + succeeded: bool, + error: Option, +} + +#[spacetimedb::reducer(on_result(self::outbound_ping))] +pub fn on_ping_result(ctx: &spacetimedb::ReducerContext, row: OutboundPing, result: Result<(), String>) { + ctx.db.ping_result().insert(PingResult { + id: row.message_number, + callback_order: 0, + module_version: MODULE_VERSION, + body: row.bodypo, + succeeded: result.is_ok(), + error: result.err(), + }); +} + +#[spacetimedb::table(accessor = outbound_detailed_ping, outbox(game_world::receive_detailed_ping))] +pub struct OutboundDetailedPing { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: game_world::Identity, + body: String, + details: game_world::PingDetails, +} + +#[spacetimedb::reducer] +pub fn enqueue_detailed_ping(ctx: &spacetimedb::ReducerContext, target: game_world::Identity, body: String) { + ctx.db.outbound_detailed_ping().insert(OutboundDetailedPing { + id: 0, + target, + body, + details: game_world::PingDetails { + labels: vec!["nested".into(), "idc".into()], + checkpoints: vec![ + game_world::PingCheckpoint { + sequence: 1, + state: game_world::PingState::Ready, + }, + game_world::PingCheckpoint { + sequence: 2, + state: game_world::PingState::Deferred(Some("waiting".into())), + }, + game_world::PingCheckpoint { + sequence: 3, + state: game_world::PingState::Deferred(None), + }, + ], + note: Some("optional-note".into()), + }, + }); +} + +#[spacetimedb::table(accessor = detailed_ping_result)] +pub struct DetailedPingResult { + #[primary_key] + id: u64, + body: String, + details: game_world::PingDetails, +} + +#[spacetimedb::reducer(on_result(self::outbound_detailed_ping))] +pub fn on_detailed_ping_result( + ctx: &spacetimedb::ReducerContext, + row: OutboundDetailedPing, + _result: Result<(), String>, +) { + ctx.db.detailed_ping_result().insert(DetailedPingResult { + id: row.id, + body: row.body, + details: row.details, + }); +} diff --git a/crates/smoketests/modules/idc-sender/src/remote_bindings/mod.rs b/crates/smoketests/modules/idc-sender/src/remote_bindings/mod.rs new file mode 100644 index 00000000000..8b137891791 --- /dev/null +++ b/crates/smoketests/modules/idc-sender/src/remote_bindings/mod.rs @@ -0,0 +1 @@ + diff --git a/crates/smoketests/modules/idc-two-dependencies/Cargo.toml b/crates/smoketests/modules/idc-two-dependencies/Cargo.toml new file mode 100644 index 00000000000..35631d6352f --- /dev/null +++ b/crates/smoketests/modules/idc-two-dependencies/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "smoketest-module-idc-two-dependencies" +version = "0.1.0" +edition = "2021" +publish = false + +[lib] +crate-type = ["cdylib"] + +[dependencies] +spacetimedb.workspace = true diff --git a/crates/smoketests/modules/idc-two-dependencies/src/lib.rs b/crates/smoketests/modules/idc-two-dependencies/src/lib.rs new file mode 100644 index 00000000000..87e7ff91206 --- /dev/null +++ b/crates/smoketests/modules/idc-two-dependencies/src/lib.rs @@ -0,0 +1,49 @@ +mod remote_bindings; + +use remote_bindings::{auth, game_world}; + +#[spacetimedb::table(accessor = world_outbox, outbox(game_world::receive_detailed_ping))] +pub struct WorldOutbox { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: game_world::Identity, + body: String, + details: game_world::PingDetails, +} + +#[spacetimedb::table(accessor = auth_outbox, outbox(auth::receive_detailed_ping))] +pub struct AuthOutbox { + #[primary_key] + #[auto_inc] + id: u64, + #[target] + target: auth::Identity, + body: String, + details: auth::PingDetails, +} + +#[spacetimedb::reducer] +pub fn enqueue_both(ctx: &spacetimedb::ReducerContext, world: game_world::Identity, auth: auth::Identity) { + ctx.db.world_outbox().insert(WorldOutbox { + id: 0, + target: world, + body: "world".into(), + details: game_world::PingDetails { + labels: vec![], + checkpoints: vec![], + note: None, + }, + }); + ctx.db.auth_outbox().insert(AuthOutbox { + id: 0, + target: auth, + body: "auth".into(), + details: auth::PingDetails { + labels: vec![], + checkpoints: vec![], + note: None, + }, + }); +} diff --git a/crates/smoketests/modules/idc-two-dependencies/src/remote_bindings/mod.rs b/crates/smoketests/modules/idc-two-dependencies/src/remote_bindings/mod.rs new file mode 100644 index 00000000000..8b137891791 --- /dev/null +++ b/crates/smoketests/modules/idc-two-dependencies/src/remote_bindings/mod.rs @@ -0,0 +1 @@ + diff --git a/crates/smoketests/src/lib.rs b/crates/smoketests/src/lib.rs index feaa1e6bf2b..2b9285dcab7 100644 --- a/crates/smoketests/src/lib.rs +++ b/crates/smoketests/src/lib.rs @@ -189,7 +189,7 @@ pub fn workspace_root() -> PathBuf { .to_path_buf() } -/// Copies a Rust module fixture's manifest and source into a temporary project. +/// Copies a Rust module fixture into a temporary project with a unique package name. pub fn copy_module_fixture(name: &str, dest: &Path) -> Result<()> { let source = workspace_root().join("crates/smoketests/modules").join(name); fs::create_dir_all(dest.join("src"))?; @@ -197,6 +197,14 @@ pub fn copy_module_fixture(name: &str, dest: &Path) -> Result<()> { fs::copy(source.join(file), dest.join(file)) .with_context(|| format!("Failed to copy {file} from fixture {name}"))?; } + let manifest_path = dest.join("Cargo.toml"); + let manifest = fs::read_to_string(&manifest_path)?; + let package_name = Regex::new(r#"(?m)^name\s*=\s*"([^"]+)""#)?; + let suffix = random_string(); + let manifest = package_name.replacen(&manifest, 1, |captures: ®ex::Captures<'_>| { + format!("name = \"{}-{suffix}\"", &captures[1]) + }); + fs::write(manifest_path, manifest.as_bytes())?; Ok(()) } @@ -466,7 +474,8 @@ if(CMAKE_SYSTEM_NAME STREQUAL "Emscripten") endif() "#; -fn parse_identity_from_publish_output(publish_output: &str) -> Result { +/// Extracts the database identity from a single database publication. +pub fn parse_identity_from_publish_output(publish_output: &str) -> Result { let re = Regex::new(r"identity: ([0-9a-fA-F]+)").unwrap(); re.captures(publish_output) .and_then(|caps| caps.get(1)) @@ -1552,7 +1561,15 @@ log = "0.4" pub fn call(&self, name: &str, args: &[&str]) -> Result { let identity = self.database_identity.as_ref().context("No database published")?; - let mut cmd_args = vec!["call", "--server", &self.server_url, "--", identity.as_str(), name]; + let mut cmd_args = vec![ + "call", + "--no-config", + "--server", + &self.server_url, + "--", + identity.as_str(), + name, + ]; cmd_args.extend(args); self.spacetime(&cmd_args) @@ -1562,7 +1579,15 @@ log = "0.4" pub fn call_output(&self, name: &str, args: &[&str]) -> Output { let identity = self.database_identity.as_ref().expect("No database published"); - let mut cmd_args = vec!["call", "--server", &self.server_url, "--", identity.as_str(), name]; + let mut cmd_args = vec![ + "call", + "--no-config", + "--server", + &self.server_url, + "--", + identity.as_str(), + name, + ]; cmd_args.extend(args); self.spacetime_cmd(&cmd_args) @@ -1574,6 +1599,7 @@ log = "0.4" let mut cmd_args = vec![ "call", + "--no-config", "--anonymous", "--server", &self.server_url, @@ -1956,6 +1982,33 @@ fn normalize_whitespace(s: &str) -> String { mod tests { use super::*; + #[test] + fn copied_module_fixtures_have_distinct_wasm_names() { + let temp = tempfile::tempdir().unwrap(); + let target_names: Vec<_> = ["first", "second"] + .into_iter() + .map(|directory| { + let module_dir = temp.path().join(directory); + copy_module_fixture("idc-receiver", &module_dir).unwrap(); + patch_module_cargo_to_local_bindings(&module_dir).unwrap(); + let metadata = cargo_metadata::MetadataCommand::new() + .manifest_path(module_dir.join("Cargo.toml")) + .no_deps() + .exec() + .unwrap(); + let package = metadata.root_package().unwrap(); + package + .targets + .iter() + .find(|target| target.crate_types.iter().any(|kind| kind == "cdylib")) + .unwrap() + .name + .clone() + }) + .collect(); + assert_ne!(target_names[0], target_names[1]); + } + #[test] fn test_normalize_whitespace() { let input = "hello \nworld \n foo "; diff --git a/crates/smoketests/tests/cluster.rs b/crates/smoketests/tests/cluster.rs index f0aa305b01a..62ca5ff4ca7 100644 --- a/crates/smoketests/tests/cluster.rs +++ b/crates/smoketests/tests/cluster.rs @@ -23,6 +23,7 @@ mod cluster { mod filtering; mod http_egress; mod http_routes; + mod idc; mod logs_level_filter; mod module_nested_op; mod modules; diff --git a/crates/smoketests/tests/cluster/idc.rs b/crates/smoketests/tests/cluster/idc.rs new file mode 100644 index 00000000000..0c1472fdf9c --- /dev/null +++ b/crates/smoketests/tests/cluster/idc.rs @@ -0,0 +1,533 @@ +use std::time::{Duration, Instant}; + +use spacetimedb_smoketests::{ + copy_module_fixture, parse_identity_from_publish_output, patch_module_cargo_to_local_bindings, random_string, + Smoketest, +}; + +struct IdcTest { + test: Smoketest, + receiver: String, + sender: String, +} + +impl IdcTest { + fn new() -> Self { + let mut test = Smoketest::builder().autopublish(false).build(); + let receiver_db = format!("idc-receiver-{}", random_string()); + let sender_db = format!("idc-sender-{}", random_string()); + + for name in ["receiver", "sender"] { + let dir = test.project_dir.path().join(name); + copy_module_fixture(&format!("idc-{name}"), &dir).expect("failed to copy module fixture"); + patch_module_cargo_to_local_bindings(&dir).expect("failed to patch module Cargo.toml"); + } + std::fs::write( + test.project_dir.path().join("spacetime.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "server": test.server_url, + "database": receiver_db, + "module-path": "./receiver", + "children": [ + { + "database": sender_db, + "module-path": "./sender", + "dependencies": [{ "name": "game_world", "database": receiver_db }] + } + ] + })) + .unwrap(), + ) + .expect("failed to write spacetime.json"); + test.spacetime(&["generate", &sender_db]) + .expect("failed to generate receiver bindings"); + test.spacetime(&["publish", "--server", &test.server_url, "--yes=all"]) + .expect("failed to publish IDC modules"); + + let resolve_identity = |database: &str| { + let response = test + .api_call("GET", &format!("/v1/database/{database}/identity")) + .expect("failed to resolve database identity"); + assert!( + response.is_success(), + "identity lookup failed for {database}: HTTP {}", + response.status_code + ); + response.text().expect("invalid database identity response") + }; + let receiver = resolve_identity(&receiver_db); + let sender = resolve_identity(&sender_db); + test.database_identity = Some(sender.clone()); + Self { test, receiver, sender } + } + + fn publish(&self, module: &str, database: &str) -> String { + let output = self + .test + .spacetime(&[ + "publish", + "--no-config", + "--server", + &self.test.server_url, + "--module-path", + self.test.project_dir.path().join(module).to_str().unwrap(), + "--yes=all", + database, + ]) + .expect("failed to publish module update"); + parse_identity_from_publish_output(&output).expect("publish output should contain database identity") + } + + fn rows(&self, database: &str, sql: &str) -> Vec { + let output = self + .test + .spacetime(&[ + "sql", + "--no-config", + "--server", + &self.test.server_url, + "--format", + "json", + database, + sql, + ]) + .expect("failed to query database"); + let results: serde_json::Value = serde_json::from_str(&output).expect("invalid SQL JSON output"); + results[0]["rows"] + .as_array() + .expect("SQL result should contain rows") + .clone() + } + + fn enqueue(&self, bodies: &[String]) { + self.test + .call( + "enqueue_pings", + &[ + &serde_json::to_string(&self.receiver).unwrap(), + &serde_json::to_string(bodies).unwrap(), + ], + ) + .expect("failed to enqueue pings"); + } + + fn wait_for_messages(&self, expected: &[(String, u32)]) { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + // The auto-increment columns record execution order, independently of message IDs. + let mut delivered = self.rows(&self.receiver, "SELECT id, body, module_version FROM ping_log"); + let mut callbacks = self.rows( + &self.sender, + "SELECT callback_order, body, module_version, succeeded, error FROM ping_result", + ); + delivered.sort_by_key(|row| row[0].as_u64().unwrap()); + callbacks.sort_by_key(|row| row[0].as_u64().unwrap()); + let pending = self.rows(&self.sender, "SELECT message_number FROM outbound_ping"); + assert!(delivered.len() <= expected.len(), "duplicate deliveries: {delivered:?}"); + assert!(callbacks.len() <= expected.len(), "duplicate callbacks: {callbacks:?}"); + if delivered.len() == expected.len() && callbacks.len() == expected.len() && pending.is_empty() { + for (index, (body, version)) in expected.iter().enumerate() { + assert_eq!(delivered[index][1], *body, "receiver delivery order"); + assert_eq!(delivered[index][2], *version, "receiver module version"); + assert_eq!(callbacks[index][1], *body, "sender callback order"); + assert_eq!(callbacks[index][2], *version, "sender module version"); + assert_eq!(callbacks[index][3], true, "callback failed: {:?}", callbacks[index]); + } + return; + } + assert!( + Instant::now() < deadline, + "IDC did not finish: delivered={delivered:?}, callbacks={callbacks:?}, pending={pending:?}" + ); + std::thread::sleep(Duration::from_millis(100)); + } + } + + fn send_ping_to(&self, receiver: &str, body: &str) { + self.test + .call( + "enqueue_ping", + &[ + &serde_json::to_string(receiver).unwrap(), + &serde_json::to_string(body).unwrap(), + ], + ) + .expect("failed to enqueue ping"); + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let callbacks = self.rows( + &self.sender, + &format!("SELECT succeeded FROM ping_result WHERE body = '{body}'"), + ); + let pending = self.rows(&self.sender, "SELECT message_number FROM outbound_ping"); + if !callbacks.is_empty() && pending.is_empty() { + assert_eq!(callbacks, vec![serde_json::json!([true])]); + assert_eq!( + self.rows(receiver, &format!("SELECT body FROM ping_log WHERE body = '{body}'")), + vec![serde_json::json!([body])], + ); + return; + } + assert!( + Instant::now() < deadline, + "ping {body} did not finish: {callbacks:?}, {pending:?}" + ); + std::thread::sleep(Duration::from_millis(100)); + } + } + + fn send_detailed_ping(&self, body: &str) { + self.test + .call( + "enqueue_detailed_ping", + &[ + &serde_json::to_string(&self.receiver).unwrap(), + &serde_json::to_string(body).unwrap(), + ], + ) + .expect("failed to enqueue detailed ping"); + let deadline = Instant::now() + Duration::from_secs(20); + let expected_details = serde_json::json!([ + ["nested", "idc"], + [[1, [0, []]], [2, [1, [0, "waiting"]]], [3, [1, [1, []]]]], + [0, "optional-note"] + ]); + loop { + let callbacks = self.rows( + &self.sender, + &format!("SELECT details FROM detailed_ping_result WHERE body = '{body}'"), + ); + let pending = self.rows(&self.sender, "SELECT id FROM outbound_detailed_ping"); + let delivered = self.rows( + &self.receiver, + &format!("SELECT details FROM detailed_ping_log WHERE body = '{body}'"), + ); + if !callbacks.is_empty() && pending.is_empty() { + assert_eq!(delivered, vec![serde_json::json!([expected_details])]); + assert_eq!(callbacks, vec![serde_json::json!([expected_details])]); + return; + } + assert!( + Instant::now() < deadline, + "detailed IDC ping did not finish: delivered={delivered:?}, callbacks={callbacks:?}, pending={pending:?}" + ); + std::thread::sleep(Duration::from_millis(100)); + } + } + + fn edit_module(&self, module: &str, from: &str, to: &str) { + let path = self.test.project_dir.path().join(module).join("src/lib.rs"); + let source = std::fs::read_to_string(&path).unwrap().replace("\r\n", "\n"); + assert!(source.contains(from), "missing source in {module}: {from}"); + std::fs::write(&path, source.replace(from, to)).unwrap(); + } + + fn update_modules(&self, version: u32) { + for (module, database) in [("receiver", &self.receiver), ("sender", &self.sender)] { + self.edit_module( + module, + &format!("const MODULE_VERSION: u32 = {};", version - 1), + &format!("const MODULE_VERSION: u32 = {version};"), + ); + assert_eq!(self.publish(module, database), *database); + } + } +} + +#[test] +fn idc_outbox_delivers_between_two_rust_modules() { + let suite = IdcTest::new(); + suite + .test + .call( + "enqueue_ping", + &[&serde_json::to_string(&suite.receiver).unwrap(), "\"hello-idc\""], + ) + .expect("failed to enqueue ping"); + suite.wait_for_messages(&[("hello-idc".into(), 1)]); +} + +#[test] +fn idc_outbox_delivers_twenty_messages_in_order() { + let suite = IdcTest::new(); + let bodies: Vec<_> = (0..20).map(|index| format!("message-{index}")).collect(); + suite.enqueue(&bodies); + suite.wait_for_messages(&bodies.into_iter().map(|body| (body, 1)).collect::>()); +} + +#[test] +fn idc_outbox_delivers_to_two_databases_with_the_same_receiver_module() { + let suite = IdcTest::new(); + let second_receiver = suite.publish("receiver", &format!("idc-receiver-{}", random_string())); + assert_ne!(suite.receiver, second_receiver); + + // The same bindings work for both databases before either receiver changes. + suite.send_ping_to(&suite.receiver, "first-receiver"); + suite.send_ping_to(&second_receiver, "second-receiver"); + + // Update only the first receiver. Existing rows get revision 0; new calls supply it. + suite.edit_module( + "receiver", + " module_version: u32,\n body: String,", + " module_version: u32,\n body: String,\n #[default(0)]\n revision: u32,", + ); + suite.edit_module( + "receiver", + "pub fn receive_ping(ctx: &spacetimedb::ReducerContext, body: String)", + "pub fn receive_ping(ctx: &spacetimedb::ReducerContext, body: String, revision: u32)", + ); + suite.edit_module("receiver", " body,\n", " body,\n revision,\n"); + assert_eq!(suite.publish("receiver", &suite.receiver), suite.receiver); + assert_eq!( + suite.rows(&suite.receiver, "SELECT body, revision FROM ping_log"), + vec![serde_json::json!(["first-receiver", 0])], + ); + + // Regenerate the first receiver's bindings and add the new argument to the sender's outbox. + suite + .test + .spacetime(&["generate"]) + .expect("failed to regenerate receiver bindings"); + suite.edit_module( + "sender", + " bodypo: String,", + " bodypo: String,\n #[default(0)]\n revision: u32,", + ); + suite.edit_module( + "sender", + " bodypo: body,", + " bodypo: body,\n revision: 7,", + ); + assert_eq!(suite.publish("sender", &suite.sender), suite.sender); + suite.send_ping_to(&suite.receiver, "after-update"); + assert_eq!( + suite.rows( + &suite.receiver, + "SELECT revision FROM ping_log WHERE body = 'after-update'" + ), + vec![serde_json::json!([7])], + ); + assert_eq!( + suite.rows(&second_receiver, "SELECT body FROM ping_log"), + vec![serde_json::json!(["second-receiver"])], + ); +} + +#[test] +fn idc_outboxes_deliver_in_both_directions_concurrently() { + let suite = IdcTest::new(); + let peer = suite.publish("sender", &format!("idc-peer-{}", random_string())); + let from_sender: Vec<_> = (0..20).map(|index| format!("sender-{index}")).collect(); + let from_peer: Vec<_> = (0..20).map(|index| format!("peer-{index}")).collect(); + + // Release both calls together, without waiting for either direction's deliveries. + let start = std::sync::Barrier::new(2); + std::thread::scope(|scope| { + for (source, target, bodies) in [(&suite.sender, &peer, &from_sender), (&peer, &suite.sender, &from_peer)] { + let (test, start) = (&suite.test, &start); + scope.spawn(move || { + start.wait(); + test.spacetime(&[ + "call", + "--no-config", + "--server", + &test.server_url, + "--", + source, + "enqueue_pings", + &serde_json::to_string(target).unwrap(), + &serde_json::to_string(bodies).unwrap(), + ]) + .expect("failed to enqueue concurrent pings"); + }); + } + }); + + for (database, incoming, outgoing) in [ + (&suite.sender, &from_peer, &from_sender), + (&peer, &from_sender, &from_peer), + ] { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let mut delivered = suite.rows(database, "SELECT id, body FROM ping_log"); + let mut callbacks = suite.rows(database, "SELECT callback_order, body, succeeded FROM ping_result"); + let pending = suite.rows(database, "SELECT message_number FROM outbound_ping"); + assert!(delivered.len() <= incoming.len(), "duplicate deliveries: {delivered:?}"); + assert!(callbacks.len() <= outgoing.len(), "duplicate callbacks: {callbacks:?}"); + if delivered.len() == incoming.len() && callbacks.len() == outgoing.len() && pending.is_empty() { + delivered.sort_by_key(|row| row[0].as_u64().unwrap()); + callbacks.sort_by_key(|row| row[0].as_u64().unwrap()); + for (row, body) in delivered.iter().zip(incoming) { + assert_eq!(row[1], *body, "delivery order for {database}"); + } + for (row, body) in callbacks.iter().zip(outgoing) { + assert_eq!(row[1], *body, "callback order for {database}"); + assert_eq!(row[2], true, "callback failed: {row:?}"); + } + break; + } + assert!( + Instant::now() < deadline, + "bidirectional IDC did not finish for {database}: delivered={delivered:?}, callbacks={callbacks:?}, pending={pending:?}" + ); + std::thread::sleep(Duration::from_millis(100)); + } + } +} + +#[test] +fn idc_outbox_preserves_order_across_module_updates() { + let suite = IdcTest::new(); + let mut expected = Vec::new(); + for version in 1..=4 { + if version == 2 { + // Rename the Rust accessor while preserving the canonical name used by the old bindings. + suite.edit_module( + "receiver", + "#[spacetimedb::reducer]\npub fn receive_ping", + "#[spacetimedb::reducer(name = \"receive_ping\")]\npub fn accept_ping", + ); + } + if version > 1 { + suite.update_modules(version); + } + let bodies: Vec<_> = (expected.len()..expected.len() + 5) + .map(|index| format!("message-{index}")) + .collect(); + suite.enqueue(&bodies); + expected.extend(bodies.into_iter().map(|body| (body, version))); + suite.wait_for_messages(&expected); + } +} + +#[test] +fn idc_outbox_uses_regenerated_bindings_after_argument_reordering() { + let suite = IdcTest::new(); + suite.send_detailed_ping("before-update"); + + // Changing parameter order changes the signature hash and the outbox's argument-column mapping. + // The stored row types stay the same, so both databases can be updated without clearing data. + suite.edit_module( + "receiver", + "body: String, details: PingDetails", + "details: PingDetails, body: String", + ); + suite + .test + .spacetime(&["generate"]) + .expect("failed to regenerate receiver bindings"); + assert_eq!(suite.publish("receiver", &suite.receiver), suite.receiver); + // The regenerated bindings map the unchanged sender columns into the new argument order. + assert_eq!(suite.publish("sender", &suite.sender), suite.sender); + + suite.send_detailed_ping("after-update"); +} + +#[test] +fn idc_generated_receiver_types_are_scoped_by_dependency() { + let mut test = Smoketest::builder().autopublish(false).build(); + let world_db = format!("idc-world-{}", random_string()); + let auth_db = format!("idc-auth-{}", random_string()); + let sender_db = format!("idc-scoped-sender-{}", random_string()); + + for (fixture, directory) in [ + ("idc-receiver", "world"), + ("idc-receiver", "auth"), + ("idc-two-dependencies", "sender"), + ] { + let module_dir = test.project_dir.path().join(directory); + copy_module_fixture(fixture, &module_dir).expect("failed to copy IDC fixture"); + patch_module_cargo_to_local_bindings(&module_dir).expect("failed to patch module Cargo.toml"); + } + std::fs::write( + test.project_dir.path().join("spacetime.json"), + serde_json::to_vec_pretty(&serde_json::json!({ + "server": test.server_url, + "database": world_db, + "module-path": "./world", + "children": [ + { "database": auth_db, "module-path": "./auth" }, + { + "database": sender_db, + "module-path": "./sender", + "dependencies": [ + { "name": "game_world", "database": world_db }, + { "name": "auth", "database": auth_db } + ] + } + ] + })) + .unwrap(), + ) + .expect("failed to write spacetime.json"); + + test.spacetime(&["generate", &sender_db]) + .expect("failed to generate both receiver bindings"); + for dependency in ["game_world", "auth"] { + let file = test + .project_dir + .path() + .join("sender/src/remote_bindings") + .join(dependency) + .join("ping_details_type.rs"); + let source = std::fs::read_to_string(file).expect("generated receiver type is missing"); + assert!(source.contains(&format!("#[sats(name = \"{dependency}.PingDetails\")]"))); + } + test.spacetime(&["publish", "--server", &test.server_url, "--yes=all"]) + .expect("failed to publish modules with same-named receiver types"); + + let resolve_identity = |database: &str| { + let response = test + .api_call("GET", &format!("/v1/database/{database}/identity")) + .expect("failed to resolve database identity"); + assert!(response.is_success(), "identity lookup failed for {database}"); + response.text().expect("invalid database identity response") + }; + let world = resolve_identity(&world_db); + let auth = resolve_identity(&auth_db); + test.database_identity = Some(resolve_identity(&sender_db)); + test.call( + "enqueue_both", + &[ + &serde_json::to_string(&world).unwrap(), + &serde_json::to_string(&auth).unwrap(), + ], + ) + .expect("failed to enqueue both remote messages"); + + let rows = |database: &str, sql: &str| { + let output = test + .spacetime(&[ + "sql", + "--no-config", + "--server", + &test.server_url, + "--format", + "json", + database, + sql, + ]) + .expect("failed to query IDC database"); + let result: serde_json::Value = serde_json::from_str(&output).expect("invalid SQL JSON output"); + result[0]["rows"].as_array().unwrap().clone() + }; + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let world_messages = rows(&world_db, "SELECT body FROM detailed_ping_log"); + let auth_messages = rows(&auth_db, "SELECT body FROM detailed_ping_log"); + let world_pending = rows(&sender_db, "SELECT id FROM world_outbox"); + let auth_pending = rows(&sender_db, "SELECT id FROM auth_outbox"); + if world_messages == vec![serde_json::json!(["world"])] + && auth_messages == vec![serde_json::json!(["auth"])] + && world_pending.is_empty() + && auth_pending.is_empty() + { + break; + } + assert!( + Instant::now() < deadline, + "scoped IDC did not finish: world={world_messages:?}, auth={auth_messages:?}, world_pending={world_pending:?}, auth_pending={auth_pending:?}" + ); + std::thread::sleep(Duration::from_millis(100)); + } +} diff --git a/crates/smoketests/tests/standalone/cli/generate.rs b/crates/smoketests/tests/standalone/cli/generate.rs index 4794fb58403..21c3f820dce 100644 --- a/crates/smoketests/tests/standalone/cli/generate.rs +++ b/crates/smoketests/tests/standalone/cli/generate.rs @@ -222,3 +222,70 @@ fn cli_generate_two_scoped_receiver_types_compile_in_sender() { String::from_utf8_lossy(&output.stderr) ); } +/// Standalone-only: a sender module can use generated receiver bindings from a spacetime.json dependency. +#[test] +fn cli_generate_receiver_bindings_from_dependency_compile_in_sender() { + let temp_dir = tempfile::tempdir().expect("failed to create temp dir"); + let receiver_dir = temp_dir.path().join("receiver"); + let sender_dir = temp_dir.path().join("sender"); + + copy_module_fixture("idc-receiver", &receiver_dir).expect("failed to copy receiver fixture"); + copy_module_fixture("idc-sender", &sender_dir).expect("failed to copy sender fixture"); + patch_module_cargo_to_local_bindings(&receiver_dir).expect("failed to patch receiver Cargo.toml"); + patch_module_cargo_to_local_bindings(&sender_dir).expect("failed to patch sender Cargo.toml"); + + std::fs::write( + temp_dir.path().join("spacetime.json"), + r#"{ + "server": "local", + "children": [ + { + "database": "game-world", + "module-path": "./receiver" + }, + { + "database": "lobby", + "module-path": "./sender", + "dependencies": [ + { "name": "game_world", "database": "game-world" } + ] + } + ] +} +"#, + ) + .expect("failed to write spacetime.json"); + + let output = cli_cmd() + .args(["generate", "lobby"]) + .current_dir(temp_dir.path()) + .output() + .expect("failed to execute"); + assert!( + output.status.success(), + "generate failed:\nstdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + let remote_bindings_mod = std::fs::read_to_string(sender_dir.join("src/remote_bindings/mod.rs")) + .expect("generated remote_bindings mod.rs should exist"); + assert!(remote_bindings_mod.contains("pub mod game_world;")); + + let generated = sender_dir.join("src/remote_bindings/game_world/mod.rs"); + let root = std::fs::read_to_string(&generated).expect("generated receiver binding should exist"); + assert!(root.contains("pub struct Identity(pub spacetimedb::Identity);")); + assert!(root.contains("impl spacetimedb::rt::RemoteReducer for receive_ping")); + assert!(root.contains("receive_ping")); + + let output = cli_cmd() + .args(["build", "--module-path", sender_dir.to_str().unwrap()]) + .output() + .expect("failed to execute"); + assert!( + output.status.success(), + "sender build failed:\nstdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +}