Merge branch 'master' into kim/snapshot/fsync

Author: kim

crates/bindings-csharp/BSATN.Runtime.Tests/Tests.cs

@@ -1,7 +1,9 @@
 namespace SpacetimeDB;
 
 using System.Diagnostics.CodeAnalysis;
+using System.Globalization;
 using System.Security.Cryptography;
+using System.Threading;
 using CsCheck;
 using SpacetimeDB.BSATN;
 using Xunit;
@@ -186,6 +188,27 @@ public static void TimestampConversionChecks()
         Assert.Equal(+1, laterStamp.CompareTo(stamp));
     }
 
+    [Fact]
+    public static void TimestampSqlLiteralUsesInvariantCulture()
+    {
+        var originalCulture = Thread.CurrentThread.CurrentCulture;
+
+        try
+        {
+            // Ensure the format is agnostic to the culture. Using ar-SA because it's different than Gregorian, which is used in UTC.
+            Thread.CurrentThread.CurrentCulture = CultureInfo.GetCultureInfo("ar-SA");
+
+            Assert.Equal(
+                "'2025-01-22T21:53:13.990639Z'",
+                SqlLit.Timestamp(new Timestamp(1_737_582_793_990_639L)).ToString()
+            );
+        }
+        finally
+        {
+            Thread.CurrentThread.CurrentCulture = originalCulture;
+        }
+    }
+
     [Fact]
     public static void ConnectionIdComparableChecks()
     {

crates/bindings-csharp/BSATN.Runtime/QueryBuilder.cs

@@ -58,6 +58,9 @@ public static SqlLiteral<ConnectionId> ConnectionId(ConnectionId value) =>
 
     public static SqlLiteral<Uuid> Uuid(Uuid value) =>
         new(SqlFormat.FormatHexLiteral(value.ToString()));
+
+    public static SqlLiteral<Timestamp> Timestamp(Timestamp value) =>
+        new(SqlFormat.FormatTimestampLiteral(value));
 }
 
 public interface IQuery<TRow>
@@ -744,6 +747,24 @@ public static BoolExpr<TRow> Eq<TRow>(this Col<TRow, Uuid> col, Uuid value) =>
     public static BoolExpr<TRow> Neq<TRow>(this Col<TRow, Uuid> col, Uuid value) =>
         col.Neq(SqlLit.Uuid(value));
 
+    public static BoolExpr<TRow> Eq<TRow>(this Col<TRow, Timestamp> col, Timestamp value) =>
+        col.Eq(SqlLit.Timestamp(value));
+
+    public static BoolExpr<TRow> Neq<TRow>(this Col<TRow, Timestamp> col, Timestamp value) =>
+        col.Neq(SqlLit.Timestamp(value));
+
+    public static BoolExpr<TRow> Lt<TRow>(this Col<TRow, Timestamp> col, Timestamp value) =>
+        col.Lt(SqlLit.Timestamp(value));
+
+    public static BoolExpr<TRow> Lte<TRow>(this Col<TRow, Timestamp> col, Timestamp value) =>
+        col.Lte(SqlLit.Timestamp(value));
+
+    public static BoolExpr<TRow> Gt<TRow>(this Col<TRow, Timestamp> col, Timestamp value) =>
+        col.Gt(SqlLit.Timestamp(value));
+
+    public static BoolExpr<TRow> Gte<TRow>(this Col<TRow, Timestamp> col, Timestamp value) =>
+        col.Gte(SqlLit.Timestamp(value));
+
     public static BoolExpr<TRow> Eq<TRow>(this IxCol<TRow, string> col, ReadOnlySpan<char> value) =>
         col.Eq(SqlLit.String(value));
 
@@ -831,10 +852,18 @@ public static BoolExpr<TRow> Eq<TRow>(this IxCol<TRow, Uuid> col, Uuid value) =>
 
     public static BoolExpr<TRow> Neq<TRow>(this IxCol<TRow, Uuid> col, Uuid value) =>
         col.Neq(SqlLit.Uuid(value));
+
+    public static BoolExpr<TRow> Eq<TRow>(this IxCol<TRow, Timestamp> col, Timestamp value) =>
+        col.Eq(SqlLit.Timestamp(value));
+
+    public static BoolExpr<TRow> Neq<TRow>(this IxCol<TRow, Timestamp> col, Timestamp value) =>
+        col.Neq(SqlLit.Timestamp(value));
 }
 
 internal static class SqlFormat
 {
+    private const string TimestampFormat = "yyyy-MM-dd'T'HH:mm:ss.ffffff'Z'";
+
     public static string QuoteIdent(string ident)
     {
         ident ??= string.Empty;
@@ -873,4 +902,12 @@ public static string FormatHexLiteral(string hex)
 
         return $"0x{s}";
     }
+
+    public static string FormatTimestampLiteral(Timestamp timestamp) =>
+        FormatStringLiteral(
+            timestamp
+                .ToStd()
+                .ToUniversalTime()
+                .ToString(TimestampFormat, CultureInfo.InvariantCulture)
+        );
 }

crates/bindings-csharp/Runtime/Internal/ITable.cs

@@ -189,7 +189,7 @@ protected static bool DoDelete(T row)
         using var stream = new MemoryStream();
         using var writer = new BinaryWriter(stream);
         // `datastore_delete_all_by_eq_bsatn` expects an array-like BSATN.
-        // Write a length of 1 without actually wrapping the `row` into array
+        // Write a length of 1 without actually wrapping the `row` into an array
         // (annoyingly, that would require passing `TRW` through a bunch of APIs).
         writer.Write(1U);
         row.WriteFields(writer);

crates/bindings-typescript/src/lib/filter.ts

@@ -1,7 +1,8 @@
 import type { RowType, UntypedTableDef } from './table';
+import { Timestamp } from './timestamp';
 import { Uuid } from './uuid';
 
-export type Value = string | number | boolean | Uuid;
+export type Value = string | number | boolean | Uuid | Timestamp;
 
 export type Expr<Column extends string> =
   | { type: 'eq'; key: Column; value: Value }
@@ -77,6 +78,13 @@ export function evaluate<Column extends string>(
         if (v instanceof Uuid && typeof expr.value === 'string') {
           return v.toString() === expr.value;
         }
+
+        if (v instanceof Timestamp) {
+          // Value of the Column and passed Value are both Timestamps so compare microseconds.
+          if (expr.value instanceof Timestamp) {
+            return v.microsSinceUnixEpoch === expr.value.microsSinceUnixEpoch;
+          }
+        }
       }
       return false;
     }
@@ -103,6 +111,9 @@ function formatValue(v: Value): string {
       if (v instanceof Uuid) {
         return `'${v.toString()}'`;
       }
+      if (v instanceof Timestamp) {
+        return `'${v.toISOString()}'`;
+      }
 
       return '';
     }

crates/bindings-typescript/tests/filter.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, it } from 'vitest';
+import { eq, evaluate, toString } from '../src/lib/filter';
+import { ModuleContext, tablesToSchema } from '../src/lib/schema';
+import { table } from '../src/lib/table';
+import { Timestamp } from '../src/lib/timestamp';
+import { t } from '../src/lib/type_builders';
+
+const peopleTableDef = table(
+  { name: 'people' },
+  {
+    createdAt: t.timestamp(),
+    id: t.u32(),
+  }
+);
+
+const schemaDef = tablesToSchema(new ModuleContext(), {
+  people: peopleTableDef,
+});
+
+describe('filter.ts timestamp support', () => {
+  it('evaluates timestamp equality by microseconds', () => {
+    const ts = Timestamp.fromDate(new Date('2024-01-01T00:00:00.123Z'));
+
+    expect(evaluate(eq('createdAt', ts), { createdAt: ts })).toBe(true);
+    expect(
+      evaluate(eq('createdAt', ts), {
+        createdAt: new Timestamp(ts.microsSinceUnixEpoch + 1n),
+      })
+    ).toBe(false);
+  });
+
+  it('renders timestamp literals as ISO strings', () => {
+    const ts = Timestamp.fromDate(new Date('2024-01-01T00:00:00.123Z'));
+
+    expect(toString(schemaDef.tables.people, eq('createdAt', ts))).toBe(
+      `createdAt = '2024-01-01T00:00:00.123000Z'`
+    );
+  });
+});

crates/bindings-typescript/tests/query.test.ts

@@ -31,6 +31,7 @@ const personTable = table(
     name: t.string(),
     age: t.u32(),
     active: t.bool(),
+    createdAt: t.timestamp(),
   }
 );
 
@@ -172,6 +173,26 @@ describe('TableScan.toSql', () => {
     );
   });
 
+  it('renders Timestamp literals as RFC3339 strings', () => {
+    const qb = makeQueryBuilder(schemaDef);
+    const timestamp = Timestamp.fromDate(new Date('2024-01-01T00:00:00.123Z'));
+    const sql = toSql(qb.person.where(row => row.createdAt.eq(timestamp)));
+
+    expect(sql).toBe(
+      `SELECT * FROM "person" WHERE "person"."createdAt" = '2024-01-01T00:00:00.123000Z'`
+    );
+  });
+
+  it('supports Timestamp comparisons in where predicates', () => {
+    const qb = makeQueryBuilder(schemaDef);
+    const timestamp = Timestamp.fromDate(new Date('2024-01-01T00:00:00.123Z'));
+    const sql = toSql(qb.person.where(row => row.createdAt.gt(timestamp)));
+
+    expect(sql).toBe(
+      `SELECT * FROM "person" WHERE "person"."createdAt" > '2024-01-01T00:00:00.123000Z'`
+    );
+  });
+
   it('renders semijoin queries without additional filters', () => {
     const qb = makeQueryBuilder(schemaDef);
     const sql = toSql(

crates/bindings/tests/ui/tables.stderr

@@ -203,10 +203,10 @@ error[E0277]: `&'a Alpha` cannot appear as an argument to an index filtering ope
   --> tests/ui/tables.rs:32:33
    |
 32 |     ctx.db.delta().compound_a().find(Alpha { beta: 0 });
-   |                                 ^^^^ should be an integer type, `bool`, `String`, `&str`, `Identity`, `Uuid`, `ConnectionId`, `Hash` or a no-payload enum which derives `SpacetimeType`, not `&'a Alpha`
+   |                                 ^^^^ should be an integer type, `bool`, `String`, `&str`, `Identity`, `Uuid`, `Timestamp`, `ConnectionId`, `Hash` or a no-payload enum which derives `SpacetimeType`, not `&'a Alpha`
    |
    = help: the trait `for<'a> FilterableValue` is not implemented for `&'a Alpha`
-   = note: The allowed set of types are limited to integers, bool, strings, `Identity`, `Uuid`, `ConnectionId`, `Hash` and no-payload enums which derive `SpacetimeType`,
+   = note: The allowed set of types are limited to integers, bool, strings, `Identity`, `Uuid`, `Timestamp`, `ConnectionId`, `Hash` and no-payload enums which derive `SpacetimeType`,
    = help: the following other types implement trait `FilterableValue`:
              &ConnectionId
              &FunctionVisibility

crates/client-api/src/routes/database.rs

@@ -9,6 +9,7 @@ use crate::auth::{
     SpacetimeIdentityToken,
 };
 use crate::routes::subscribe::generate_random_connection_id;
+use crate::util::serde::humantime_duration;
 pub use crate::util::{ByteStringBody, NameOrIdentity};
 use crate::{
     log_and_500, Action, Authorization, ControlStateDelegate, DatabaseDef, DatabaseResetDef, Host, MaybeMisdirected,
@@ -20,13 +21,14 @@ use axum::response::{ErrorResponse, IntoResponse};
 use axum::routing::MethodRouter;
 use axum::Extension;
 use axum_extra::TypedHeader;
+use derive_more::From;
 use futures::TryStreamExt;
 use http::StatusCode;
 use log::{info, warn};
 use serde::Deserialize;
 use spacetimedb::auth::identity::ConnectionAuthCtx;
 use spacetimedb::database_logger::DatabaseLogger;
-use spacetimedb::host::module_host::ClientConnectedError;
+use spacetimedb::host::module_host::{ClientConnectedError, DurabilityExited};
 use spacetimedb::host::{CallResult, UpdateDatabaseResult};
 use spacetimedb::host::{FunctionArgs, MigratePlanResult};
 use spacetimedb::host::{ModuleHost, ReducerOutcome};
@@ -44,6 +46,9 @@ use spacetimedb_lib::{sats, AlgebraicValue, Hash, ProductValue, Timestamp};
 use spacetimedb_schema::auto_migrate::{
     MigrationPolicy as SchemaMigrationPolicy, MigrationToken, PrettyPrintStyle as AutoMigratePrettyPrintStyle,
 };
+use tokio::sync::oneshot;
+use tokio::time::error::Elapsed;
+use tokio::time::timeout;
 
 use super::subscribe::{handle_websocket, HasWebSocketOptions};
 
@@ -695,8 +700,33 @@ pub struct PublishDatabaseQueryParams {
     parent: Option<NameOrIdentity>,
     #[serde(alias = "org")]
     organization: Option<NameOrIdentity>,
+    /// Duration to wait for a database update to become confirmed (i.e. durable).
+    ///
+    /// The value is parsed via the `humantime` crate, e.g. "1m", "23s", "5min".
+    ///
+    /// If not given, defaults to [default_update_confirmation_timeout].
+    /// The maximum timeout is capped by [MAX_UPDATE_CONFIRMATION_TIMEOUT].
+    ///
+    /// The parameter has no effect when creating a new database.
+    #[serde(with = "humantime_duration", default = "default_update_confirmation_timeout")]
+    update_confirmation_timeout: Duration,
+}
+
+/// Default timeout for a database update to become confirmed / durable.
+///
+/// Currently, the value is 5s.
+const fn default_update_confirmation_timeout() -> Duration {
+    Duration::from_secs(5)
 }
 
+/// Maximum timeout for a database update to become confirmed / durable.
+///
+/// If a replication group doesn't converge within this time span, it is
+/// probably not making progress at all.
+///
+/// Currently, the value is 5min.
+const MAX_UPDATE_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(5 * 60);
+
 pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
     Path(PublishDatabaseParams { name_or_identity }): Path<PublishDatabaseParams>,
@@ -708,6 +738,7 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
         host_type,
         parent,
         organization,
+        update_confirmation_timeout: confirmation_timeout,
     }): Query<PublishDatabaseQueryParams>,
     Extension(auth): Extension<SpacetimeAuth>,
     program_bytes: Bytes,
@@ -823,23 +854,72 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
         .await
         .map_err(log_and_500)?;
 
+    let success = || {
+        axum::Json(PublishResult::Success {
+            domain: db_name.cloned(),
+            database_identity,
+            op: publish_op,
+        })
+    };
     match maybe_updated {
         Some(UpdateDatabaseResult::AutoMigrateError(errs)) => {
             Err(bad_request(format!("Database update rejected: {errs}").into()))
         }
         Some(UpdateDatabaseResult::ErrorExecutingMigration(err)) => Err(bad_request(
             format!("Failed to create or update the database: {err}").into(),
         )),
-        None
-        | Some(
-            UpdateDatabaseResult::NoUpdateNeeded
-            | UpdateDatabaseResult::UpdatePerformed
-            | UpdateDatabaseResult::UpdatePerformedWithClientDisconnect,
-        ) => Ok(axum::Json(PublishResult::Success {
-            domain: db_name.cloned(),
-            database_identity,
-            op: publish_op,
-        })),
+        None | Some(UpdateDatabaseResult::NoUpdateNeeded) => Ok(success()),
+        Some(
+            UpdateDatabaseResult::UpdatePerformed {
+                tx_offset,
+                durable_offset,
+            }
+            | UpdateDatabaseResult::UpdatePerformedWithClientDisconnect {
+                tx_offset,
+                durable_offset,
+            },
+        ) => {
+            timeout(confirmation_timeout.min(MAX_UPDATE_CONFIRMATION_TIMEOUT), async {
+                let tx_offset = tx_offset.await?;
+                if let Some(mut durable_offset) = durable_offset {
+                    durable_offset.wait_for(tx_offset).await?;
+                }
+
+                Ok::<_, UpdateConfirmationError>(())
+            })
+            .await
+            .map_err(Into::into)
+            .flatten()?;
+
+            Ok(success())
+        }
+    }
+}
+
+#[derive(From)]
+enum UpdateConfirmationError {
+    Cancelled(oneshot::error::RecvError),
+    Crashed(DurabilityExited),
+    Timeout(Elapsed),
+}
+
+impl From<UpdateConfirmationError> for ErrorResponse {
+    fn from(e: UpdateConfirmationError) -> Self {
+        match e {
+            UpdateConfirmationError::Cancelled(_) => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "Database update failed: transaction was cancelled",
+            ),
+            UpdateConfirmationError::Crashed(_) => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "Database update failed: database crashed while waiting for transaction confirmation",
+            ),
+            UpdateConfirmationError::Timeout(_) => (
+                StatusCode::GATEWAY_TIMEOUT,
+                "Database update failed: timeout waiting for transaction confirmation",
+            ),
+        }
+        .into()
     }
 }