Add address: Address to ReducerContext (#299)

* Add `address: Address` to `ReducerContext`

Initial support for identifying connections via an `Address`,
passed as the caller address in `ReducerContext`.

Notable design choices:

- The same type `Address` is used for clients and dbs,
  as opposed to having distinct `DbAddress` and `ClientAddress` types.
- For automatic reducers (init, scheduled, &c), the passed `Address`
  is the database's address, not the address of the client which called `publish`.
- Clients may specify an `Address` as a query parameter
  to the `subscribe` and `call` endpoints.
  If they do not, an `Address` will be randomly generated for them.

Still to do:

- Send the client's `Address` alongside their `Identity` and token
  upon WebSocket init in `IdentityToken`,
  so the client can save its `Address` for re-connection.
- SDK support.
- C# module support.
- Documentation.
- Refuse new connections if an existing connection
  with the same `(Identity, Address)` pair exists.
- Ensure we can store `Address` in tables,
  so modules can track connected clients,
  and ser/de `Address`, so those tables can be synced to clients.

* Use `None` in `ReducerContext` for HTTP calls without client address

* Fix typo in trait argument name

* `Address` representation amenable to SDK usage

Similar to `Identity`, make `Address` a product type with a double-underscored field name.

This will allow SDKs to distinguish `Address` from `Vec<u8>`,
but does necessitate some ugly `AddressForUrl` hackery
to get ser/de working correctly in different contexts.

This commit does not yet include SDK support for `Address`,
only the server-side machinery necessary for it.

* Add client_address parameter to /publish; pass to init and update reducers

Per discussion with Kim, it's useful for SpacetimeDB-cloud
for the client_address passed to the init reducer
to be that of the caller of /database/publish,
rather than the database's address.

This commit implements that behavior. Now:
- `init` and `update` take the client_address passed to publish, if any,
  or `None` if no client_address was supplied.
- Scheduled reducers never receive a client_address.
- Reducers invoked by HTTP /call take the client_address passed to /call, if any,
  or `None` if no client_address was supplied.
- Reducers invoked by WebSocket always receive the address of the WebSocket connection.

* `Address` support in Rust client SDK

* Add test that addresses are stable within a client process

* Run rustfmt against merge changes

* Not sure why my local `cargo fmt` didn't get this...

* Add caller_address argument to reducer arguments in Rust SDK

* rustfmt again...

* Add caller address to `EventJson` message

* Python codegen for client addresses

* C# module support for client addresses

* Fix scoping error

* Add `Address`-related stuff to C# codegen

- Emit `SpacetimeDB.Address` for address types
- Add `callerAddress` field to `ReducerEvent`

* TypeScript codegen changes

* Fix merge conflict with new test

* Run rustfmt

---------

Co-authored-by: Ingvar Stepanyan <me@rreverser.com>

crates/bindings-csharp/Runtime/Module.cs

@@ -189,6 +189,7 @@ private static byte[] DescribeModule()
     private static string? CallReducer(
         uint id,
         byte[] sender_identity,
+        byte[] sender_address,
         ulong timestamp,
         byte[] args
     )
@@ -197,7 +198,7 @@ byte[] args
         {
             using var stream = new MemoryStream(args);
             using var reader = new BinaryReader(stream);
-            reducers[(int)id].Invoke(reader, new(sender_identity, timestamp));
+            reducers[(int)id].Invoke(reader, new(sender_identity, sender_address, timestamp));
             if (stream.Position != stream.Length)
             {
                 throw new Exception("Unrecognised extra bytes in the reducer arguments");

crates/bindings-csharp/Runtime/Runtime.cs

@@ -216,14 +216,61 @@ public override int GetHashCode() =>
         public static SpacetimeDB.SATS.TypeInfo<Identity> GetSatsTypeInfo() => satsTypeInfo;
     }
 
+    public struct Address : IEquatable<Address>
+    {
+        private readonly byte[] bytes;
+
+        public Address(byte[] bytes) => this.bytes = bytes;
+
+        public static readonly Address Zero = new(new byte[16]);
+
+        public bool Equals(Address other) =>
+            StructuralComparisons.StructuralEqualityComparer.Equals(bytes, other.bytes);
+
+        public override bool Equals(object? obj) => obj is Address other && Equals(other);
+
+        public static bool operator ==(Address left, Address right) => left.Equals(right);
+
+        public static bool operator !=(Address left, Address right) => !left.Equals(right);
+
+        public override int GetHashCode() =>
+            StructuralComparisons.StructuralEqualityComparer.GetHashCode(bytes);
+
+        public override string ToString() => BitConverter.ToString(bytes);
+
+        private static SpacetimeDB.SATS.TypeInfo<Address> satsTypeInfo =
+            new(
+                // We need to set type info to inlined address type as `generate` CLI currently can't recognise type references for built-ins.
+                new SpacetimeDB.SATS.ProductType
+                {
+                  { "__address_bytes", SpacetimeDB.SATS.BuiltinType.BytesTypeInfo.AlgebraicType }
+                },
+                // Concern: We use this "packed" representation (as Bytes)
+                //          in the caller_id field of reducer arguments,
+                //          but in table rows,
+                //          we send the "unpacked" representation as a product value.
+                //          It's possible that these happen to be identical
+                //          because BSATN is minimally self-describing,
+                //          but that doesn't seem like something we should count on.
+                reader => new(SpacetimeDB.SATS.BuiltinType.BytesTypeInfo.Read(reader)),
+                (writer, value) =>
+                    SpacetimeDB.SATS.BuiltinType.BytesTypeInfo.Write(writer, value.bytes)
+            );
+
+        public static SpacetimeDB.SATS.TypeInfo<Address> GetSatsTypeInfo() => satsTypeInfo;
+    }
+
     public class DbEventArgs : EventArgs
     {
         public readonly Identity Sender;
         public readonly DateTimeOffset Time;
+        public readonly Address? Address;
 
-        public DbEventArgs(byte[] senderIdentity, ulong timestamp_us)
+        public DbEventArgs(byte[] senderIdentity, byte[] senderAddress, ulong timestamp_us)
         {
             Sender = new Identity(senderIdentity);
+            var addr = new Address(senderAddress);
+            Address = addr == Runtime.Address.Zero ? null : addr;
             // timestamp is in microseconds; the easiest way to convert those w/o losing precision is to get Unix origin and add ticks which are 0.1ms each.
             Time = DateTimeOffset.UnixEpoch.AddTicks(10 * (long)timestamp_us);
         }
@@ -233,11 +280,11 @@ public DbEventArgs(byte[] senderIdentity, ulong timestamp_us)
     public static event Action<DbEventArgs>? OnDisconnect;
 
     // Note: this is accessed by C bindings.
-    private static string? IdentityConnected(byte[] sender_identity, ulong timestamp)
+    private static string? IdentityConnected(byte[] sender_identity, byte[] sender_address, ulong timestamp)
     {
         try
         {
-            OnConnect?.Invoke(new(sender_identity, timestamp));
+            OnConnect?.Invoke(new(sender_identity, sender_address, timestamp));
             return null;
         }
         catch (Exception e)
@@ -247,11 +294,11 @@ public DbEventArgs(byte[] senderIdentity, ulong timestamp_us)
     }
 
     // Note: this is accessed by C bindings.
-    private static string? IdentityDisconnected(byte[] sender_identity, ulong timestamp)
+    private static string? IdentityDisconnected(byte[] sender_identity, byte[] sender_address, ulong timestamp)
     {
         try
         {
-            OnDisconnect?.Invoke(new(sender_identity, timestamp));
+            OnDisconnect?.Invoke(new(sender_identity, sender_address, timestamp));
             return null;
         }
         catch (Exception e)

crates/bindings-csharp/Runtime/bindings.c

@@ -434,33 +434,37 @@ static Buffer return_result_buf(MonoObject* str) {
 
 __attribute__((export_name("__call_reducer__"))) Buffer __call_reducer__(
     uint32_t id,
-    Buffer sender_,
+    Buffer sender_id_,
+    Buffer sender_address_,
     uint64_t timestamp,
     Buffer args_) {
-  MonoArray* sender = stdb_buffer_consume(sender_);
+  MonoArray* sender_id = stdb_buffer_consume(sender_id_);
+  MonoArray* sender_address = stdb_buffer_consume(sender_address_);
   MonoArray* args = stdb_buffer_consume(args_);
 
   return return_result_buf(INVOKE_DOTNET_METHOD(
       "SpacetimeDB.Runtime.dll", "SpacetimeDB.Module", "FFI", "CallReducer",
-      NULL, &id, sender, &timestamp, args));
+      NULL, &id, sender_id, sender_address, &timestamp, args));
 }
 
 __attribute__((export_name("__identity_connected__"))) Buffer
-__identity_connected__(Buffer sender_, uint64_t timestamp) {
-  MonoArray* sender = stdb_buffer_consume(sender_);
+__identity_connected__(Buffer sender_id_, Buffer sender_address_, uint64_t timestamp) {
+  MonoArray* sender_id = stdb_buffer_consume(sender_id_);
+  MonoArray* sender_address = stdb_buffer_consume(sender_address_);
 
   return return_result_buf(
       INVOKE_DOTNET_METHOD("SpacetimeDB.Runtime.dll", "SpacetimeDB", "Runtime",
-                           "IdentityConnected", NULL, sender, &timestamp));
+                           "IdentityConnected", NULL, sender_id, sender_address, &timestamp));
 }
 
 __attribute__((export_name("__identity_disconnected__"))) Buffer
-__identity_disconnected__(Buffer sender_, uint64_t timestamp) {
-  MonoArray* sender = stdb_buffer_consume(sender_);
+__identity_disconnected__(Buffer sender_id_, Buffer sender_address_, uint64_t timestamp) {
+  MonoArray* sender_id = stdb_buffer_consume(sender_id_);
+  MonoArray* sender_address = stdb_buffer_consume(sender_address_);
 
   return return_result_buf(
       INVOKE_DOTNET_METHOD("SpacetimeDB.Runtime.dll", "SpacetimeDB", "Runtime",
-                           "IdentityDisconnected", NULL, sender, &timestamp));
+                           "IdentityDisconnected", NULL, sender_id, sender_address, &timestamp));
 }
 
 // Shims to avoid dependency on WASI in the generated Wasm file.

crates/bindings-macro/src/lib.rs

@@ -368,10 +368,23 @@ fn gen_reducer(original_function: ItemFn, reducer_name: &str, extra: ReducerExtr
     };
 
     let generated_function = quote! {
-        fn __reducer(__sender: spacetimedb::sys::Buffer, __timestamp: u64, __args: &[u8]) -> spacetimedb::sys::Buffer {
+        // NOTE: double-underscoring names here is unnecessary, as Rust macros are hygienic.
+        fn __reducer(
+            __sender: spacetimedb::sys::Buffer,
+            __caller_address: spacetimedb::sys::Buffer,
+            __timestamp: u64,
+            __args: &[u8]
+        ) -> spacetimedb::sys::Buffer {
             #(spacetimedb::rt::assert_reducerarg::<#arg_tys>();)*
             #(spacetimedb::rt::assert_reducerret::<#ret_ty>();)*
-            spacetimedb::rt::invoke_reducer(#func_name, __sender, __timestamp, __args, |_res| { #epilogue })
+            spacetimedb::rt::invoke_reducer(
+                #func_name,
+                __sender,
+                __caller_address,
+                __timestamp,
+                __args,
+                |_res| { #epilogue },
+            )
         }
     };
 
@@ -889,8 +902,12 @@ fn spacetimedb_connect_disconnect(item: TokenStream, connect: bool) -> syn::Resu
     let emission = quote! {
         const _: () = {
             #[export_name = #connect_disconnect_symbol]
-            extern "C" fn __connect_disconnect(__sender: spacetimedb::sys::Buffer, __timestamp: u64) -> spacetimedb::sys::Buffer {
-                spacetimedb::rt::invoke_connection_func(#func_name, __sender, __timestamp)
+            extern "C" fn __connect_disconnect(
+                __sender: spacetimedb::sys::Buffer,
+                __caller_address: spacetimedb::sys::Buffer,
+                __timestamp: u64,
+            ) -> spacetimedb::sys::Buffer {
+                spacetimedb::rt::invoke_connection_func(#func_name, __sender, __caller_address, __timestamp)
             }
         };
 

crates/bindings/src/lib.rs

@@ -23,6 +23,7 @@ pub use spacetimedb_bindings_macro::{duration, query, spacetimedb, TableType};
 pub use sats::SpacetimeType;
 pub use spacetimedb_lib;
 pub use spacetimedb_lib::sats;
+pub use spacetimedb_lib::Address;
 pub use spacetimedb_lib::AlgebraicValue;
 pub use spacetimedb_lib::Identity;
 pub use timestamp::Timestamp;
@@ -51,6 +52,14 @@ pub struct ReducerContext {
     pub sender: Identity,
     /// The time at which the reducer was started.
     pub timestamp: Timestamp,
+    /// The `Address` of the client that invoked the reducer.
+    ///
+    /// `None` if no `Address` was supplied to the `/database/call` HTTP endpoint,
+    /// or via the CLI's `spacetime call` subcommand.
+    ///
+    /// For automatic reducers, i.e. `init`, `update` and scheduled reducers,
+    /// this will be the module's `Address`.
+    pub address: Option<Address>,
 }
 
 impl ReducerContext {
@@ -59,6 +68,7 @@ impl ReducerContext {
         Self {
             sender: Identity::__dummy(),
             timestamp: Timestamp::UNIX_EPOCH,
+            address: None,
         }
     }
 }

crates/bindings/src/rt.rs

@@ -14,7 +14,7 @@ use spacetimedb_lib::de::{self, Deserialize, SeqProductAccess};
 use spacetimedb_lib::sats::typespace::TypespaceBuilder;
 use spacetimedb_lib::sats::{impl_deserialize, impl_serialize, AlgebraicType, AlgebraicTypeRef, ProductTypeElement};
 use spacetimedb_lib::ser::{Serialize, SerializeSeqProduct};
-use spacetimedb_lib::{bsatn, Identity, MiscModuleExport, ModuleDef, ReducerDef, TableDef, TypeAlias};
+use spacetimedb_lib::{bsatn, Address, Identity, MiscModuleExport, ModuleDef, ReducerDef, TableDef, TypeAlias};
 use sys::Buffer;
 
 pub use once_cell::sync::{Lazy, OnceCell};
@@ -28,11 +28,12 @@ pub use once_cell::sync::{Lazy, OnceCell};
 pub fn invoke_reducer<'a, A: Args<'a>, T>(
     reducer: impl Reducer<'a, A, T>,
     sender: Buffer,
+    client_address: Buffer,
     timestamp: u64,
     args: &'a [u8],
     epilogue: impl FnOnce(Result<(), &str>),
 ) -> Buffer {
-    let ctx = assemble_context(sender, timestamp);
+    let ctx = assemble_context(sender, timestamp, client_address);
 
     // Deserialize the arguments from a bsatn encoding.
     let SerDeArgs(args) = bsatn::from_slice(args).expect("unable to decode args");
@@ -71,21 +72,42 @@ pub fn create_index(index_name: &str, table_id: u32, index_type: sys::raw::Index
 pub fn invoke_connection_func<R: ReducerResult>(
     f: impl Fn(ReducerContext) -> R,
     sender: Buffer,
+    client_address: Buffer,
     timestamp: u64,
 ) -> Buffer {
-    let ctx = assemble_context(sender, timestamp);
+    let ctx = assemble_context(sender, timestamp, client_address);
 
     let res = with_timestamp_set(ctx.timestamp, || f(ctx).into_result());
     cvt_result(res)
 }
 
-/// Creates a reducer context from the given `sender` and `timestamp`.
-fn assemble_context(sender: Buffer, timestamp: u64) -> ReducerContext {
+/// Creates a reducer context from the given `sender`, `timestamp` and `client_address`.
+///
+/// `sender` must contain 32 bytes, from which we will read an `Identity`.
+///
+/// `timestamp` is a count of microseconds since the Unix epoch.
+///
+/// `client_address` must contain 16 bytes, from which we will read an `Address`.
+/// The all-zeros `client_address` (constructed by [`Address::__dummy`]) is used as a sentinel,
+/// and translated to `None`.
+fn assemble_context(sender: Buffer, timestamp: u64, client_address: Buffer) -> ReducerContext {
     let sender = Identity::from_byte_array(sender.read_array::<32>());
 
     let timestamp = Timestamp::UNIX_EPOCH + Duration::from_micros(timestamp);
 
-    ReducerContext { sender, timestamp }
+    let address = Address::from_arr(&client_address.read_array::<16>());
+
+    let address = if address == Address::__dummy() {
+        None
+    } else {
+        Some(address)
+    };
+
+    ReducerContext {
+        sender,
+        timestamp,
+        address,
+    }
 }
 
 /// Converts `errno` into a string message.
@@ -471,7 +493,7 @@ impl TypespaceBuilder for ModuleBuilder {
 static DESCRIBERS: Mutex<Vec<fn(&mut ModuleBuilder)>> = Mutex::new(Vec::new());
 
 /// A reducer function takes in `(Sender, Timestamp, Args)` and writes to a new `Buffer`.
-pub type ReducerFn = fn(Buffer, u64, &[u8]) -> Buffer;
+pub type ReducerFn = fn(Buffer, Buffer, u64, &[u8]) -> Buffer;
 static REDUCERS: OnceCell<Vec<ReducerFn>> = OnceCell::new();
 
 /// Describes the module into a serialized form that is returned and writes the set of `REDUCERS`.
@@ -497,8 +519,14 @@ extern "C" fn __describe_module__() -> Buffer {
 ///
 /// The result of the reducer is written into a fresh buffer.
 #[no_mangle]
-extern "C" fn __call_reducer__(id: usize, sender: Buffer, timestamp: u64, args: Buffer) -> Buffer {
+extern "C" fn __call_reducer__(
+    id: usize,
+    sender: Buffer,
+    caller_address: Buffer,
+    timestamp: u64,
+    args: Buffer,
+) -> Buffer {
     let reducers = REDUCERS.get().unwrap();
     let args = args.read();
-    reducers[id](sender, timestamp, &args)
+    reducers[id](sender, caller_address, timestamp, &args)
 }