diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java
index c031d36dd..a884c3fe9 100644
--- a/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java
+++ b/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java
@@ -1375,7 +1375,7 @@ public static void transitionDelayedWorkflows(DbContext ctx) throws SQLException
    * <p>A row that a dequeue or a peer's sweep holds is skipped and left for the next sweep. So, on
    * CockroachDB, is a row that a transaction which just committed wrote or locked, until its locks
    * are cleaned up a moment later. The status values are literals rather than parameters, so the
-   * planner can match idx_workflow_status_deadline's predicate under a generic plan.
+   * planner can match idx_workflow_status_deadline_v2's predicate under a generic plan.
    *
    * @return the IDs of the workflows cancelled
    */
diff --git a/transact/src/main/java/dev/dbos/transact/migrations/MigrationManager.java b/transact/src/main/java/dev/dbos/transact/migrations/MigrationManager.java
index d48e5f74b..8c1b60aee 100644
--- a/transact/src/main/java/dev/dbos/transact/migrations/MigrationManager.java
+++ b/transact/src/main/java/dev/dbos/transact/migrations/MigrationManager.java
@@ -26,7 +26,7 @@ public class MigrationManager {
   private static final Set<Integer> ONLINE_MIGRATIONS =
       Set.of(
           22, 23, 24, 25, 26, 27, 29, 30, 31, 32, 34, 35, 37, 45, 46, 47, 107, 111, 114, 115, 116,
-          117, 118, 119, 120, 122, 124, 125);
+          117, 118, 119, 120, 122, 124, 125, 127, 128, 129);
 
   // From this index on, every SDK defines the same migration at the same index, so a migration
   // added here must be added to all of them.
@@ -580,7 +580,12 @@ public static List<String> getMigrations(
             migration122(isCockroach),
             MIGRATION_123,
             migration124(isCockroach),
-            migration125(isCockroach)));
+            migration125(isCockroach),
+            MIGRATION_126,
+            migration127(isCockroach),
+            migration128(isCockroach),
+            migration129(isCockroach),
+            migration130(isCockroach)));
     return migrations.stream().map(m -> m.formatted(schema)).toList();
   }
 
@@ -1780,4 +1785,93 @@ static String migration125(boolean isCockroach) {
         + " IF NOT EXISTS \"idx_application_versions_owner_timestamp\""
         + " ON \"%1$s\".\"application_versions\" (\"application_name\", \"version_timestamp\" DESC)";
   }
+
+  // Migration 126: the topic a SUSPENDED recv waits on, so a send on that topic can wake it.
+  static final String MIGRATION_126 =
+      """
+      ALTER TABLE "%1$s"."workflow_status"
+          ADD COLUMN IF NOT EXISTS "suspend_key" TEXT DEFAULT NULL;
+      """;
+
+  // Migration 127: index suspended workflows by wake-up time, so the wake-up sweep reads only the
+  // ones that are due.
+  static String migration127(boolean isCockroach) {
+    return "CREATE INDEX "
+        + concurrently(isCockroach)
+        + " IF NOT EXISTS \"idx_workflow_status_suspended\""
+        + " ON \"%1$s\".\"workflow_status\" (\"delay_until_epoch_ms\")"
+        + " WHERE \"status\" = 'SUSPENDED'";
+  }
+
+  // Migration 128: idx_workflow_status_deadline (migration 122), with a predicate that also covers
+  // SUSPENDED workflows, so the workflow-timeout sweep can cancel them too.
+  static String migration128(boolean isCockroach) {
+    return "CREATE INDEX "
+        + concurrently(isCockroach)
+        + " IF NOT EXISTS \"idx_workflow_status_deadline_v2\""
+        + " ON \"%1$s\".\"workflow_status\" (\"workflow_deadline_epoch_ms\")"
+        + " WHERE \"status\" IN ('ENQUEUED', 'PENDING', 'DELAYED', 'SUSPENDED')"
+        + " AND \"workflow_deadline_epoch_ms\" IS NOT NULL";
+  }
+
+  // Migration 129: drop idx_workflow_status_deadline (migration 122), superseded by
+  // idx_workflow_status_deadline_v2 (migration 128).
+  static String migration129(boolean isCockroach) {
+    return "DROP INDEX "
+        + concurrently(isCockroach)
+        + " IF EXISTS \"%1$s\".\"idx_workflow_status_deadline\"";
+  }
+
+  // Migration 130: send_message also wakes a SUSPENDED recv waiting on its topic, as every SDK's
+  // send does.
+  //
+  // The function is dropped rather than replaced. Migration 14 here named the fourth parameter
+  // idempotency_key, where every other DBOS SDK names it message_id, and CREATE OR REPLACE cannot
+  // rename an input parameter. Dropping it first lets this migration run on a system database
+  // created by any SDK, and leaves the parameter named message_id on all of them.
+  static String migration130(boolean isCockroach) {
+    var migration =
+        """
+        DROP FUNCTION IF EXISTS "%1$s".send_message(TEXT, JSON, TEXT, TEXT);
+
+        CREATE FUNCTION "%1$s".send_message(
+            destination_id TEXT,
+            message JSON,
+            topic TEXT DEFAULT NULL,
+            message_id TEXT DEFAULT NULL
+        ) RETURNS VOID AS $$
+        DECLARE
+            v_topic TEXT := COALESCE(topic, '__null__topic__');
+            v_message_id TEXT := COALESCE(message_id, gen_random_uuid()::TEXT);
+        BEGIN
+            INSERT INTO "%1$s".notifications (
+                destination_uuid, topic, message, message_uuid, serialization
+            ) VALUES (
+                destination_id, v_topic, message, v_message_id, 'portable_json'
+            )
+            ON CONFLICT (message_uuid) DO NOTHING;
+            UPDATE "%1$s".workflow_status
+            -- Status is tested in SET, not WHERE, so the planner cannot scan idx_workflow_status_suspended.
+            SET delay_until_epoch_ms = CASE WHEN status = 'SUSPENDED' THEN 0 ELSE delay_until_epoch_ms END
+            WHERE workflow_uuid = destination_id
+                AND suspend_key = v_topic;
+        EXCEPTION
+            WHEN foreign_key_violation THEN
+                RAISE EXCEPTION 'DBOS non-existent workflow'
+                   USING DETAIL = format('Destination workflow %%s does not exist', destination_id),
+                        ERRCODE = 'foreign_key_violation';
+        END;
+        $$ LANGUAGE plpgsql;
+        """;
+    if (!isCockroach) {
+      migration +=
+          """
+
+          ALTER FUNCTION "%1$s".send_message(
+              TEXT, JSON, TEXT, TEXT
+          ) SET search_path = pg_catalog, pg_temp;
+          """;
+    }
+    return migration;
+  }
 }
diff --git a/transact/src/test/java/dev/dbos/transact/migrations/MigrationManagerTest.java b/transact/src/test/java/dev/dbos/transact/migrations/MigrationManagerTest.java
index 1617ca080..ff14632bf 100644
--- a/transact/src/test/java/dev/dbos/transact/migrations/MigrationManagerTest.java
+++ b/transact/src/test/java/dev/dbos/transact/migrations/MigrationManagerTest.java
@@ -148,6 +148,7 @@ void testRunMigrations_CreatesTables() throws Exception {
                   Constants.DB_SCHEMA, true, PgContainer.USE_COCKROACH_DB));
       var version = getVersion(conn);
       assertEquals(migrations.size(), version);
+      assertSuspendModeSchema(conn);
     }
   }
 
@@ -595,7 +596,7 @@ void testSharedMigrationsUpgradeADatabaseAtTheEndOfThisLanguagesHistory() throws
 
     var schema = Constants.DB_SCHEMA;
     var latest = MigrationManager.getMigrations(schema, true, PgContainer.USE_COCKROACH_DB).size();
-    assertEquals(125, latest, "The shared history currently ends at migration 125");
+    assertEquals(130, latest, "The shared history currently ends at migration 130");
 
     // A database last migrated by a build that predates the shared base: the runner must walk the
     // padding between this language's own history and SHARED_MIGRATION_BASE without stalling.
@@ -635,7 +636,7 @@ void testSharedMigrationsUpgradeADatabaseAtTheEndOfThisLanguagesHistory() throws
       assertIndexExists(conn, "idx_workflow_topic");
 
       assertColumnExists(conn, "notifications", "consumed_by_function_id");
-      assertIndexExists(conn, "idx_workflow_status_deadline");
+      assertSuspendModeSchema(conn);
 
       for (var index : REBUILT_INDEXES) {
         assertIndexExists(conn, index[1]);
@@ -783,13 +784,19 @@ void testMigration122_IndexesActiveDeadlines() throws Exception {
       MigrationManager.runDbosMigrations(conn, schema, migrations.subList(0, 121));
       assertEquals(121, getVersion(conn));
       assertIndexAbsent(conn, "idx_workflow_status_deadline");
+
+      MigrationManager.runDbosMigrations(conn, schema, migrations.subList(0, 122));
+      assertEquals(122, getVersion(conn));
+      assertIndexExists(conn, "idx_workflow_status_deadline");
     }
 
     MigrationManager.runMigrations(dbosConfig);
 
+    // Migration 129 drops it again, superseded by idx_workflow_status_deadline_v2 (migration 128).
     try (var conn = dataSource.getConnection()) {
       assertEquals(migrations.size(), getVersion(conn));
-      assertIndexExists(conn, "idx_workflow_status_deadline");
+      assertIndexAbsent(conn, "idx_workflow_status_deadline");
+      assertIndexExists(conn, "idx_workflow_status_deadline_v2");
     }
 
     // Re-running is a no-op.
@@ -826,6 +833,242 @@ void testMigration123_AddsCreatorXid() throws Exception {
     assertDoesNotThrow(() -> MigrationManager.runMigrations(dbosConfig));
   }
 
+  /**
+   * Migrations 126 to 130 take a database at 125 to suspend mode, whichever SDK created it. Java's
+   * migration 14 names send_message's fourth parameter idempotency_key and every other SDK's names
+   * it message_id; 130 must succeed on both and leave message_id behind.
+   */
+  @ParameterizedTest
+  @ValueSource(strings = {"idempotency_key", "message_id"})
+  void testMigrations126To130_SuspendMode(String fourthParameter) throws Exception {
+    var schema = Constants.DB_SCHEMA;
+    var dbosConfig = pgContainer.dbosConfig();
+    var useListenNotify = !PgContainer.USE_COCKROACH_DB;
+    var migrations =
+        MigrationManager.getMigrations(schema, useListenNotify, PgContainer.USE_COCKROACH_DB);
+
+    // A database last migrated by an SDK that stopped at 125.
+    MigrationManager.createDatabaseIfNotExists(
+        pgContainer.jdbcUrl(), pgContainer.username(), pgContainer.password());
+    try (var conn = dataSource.getConnection()) {
+      MigrationManager.ensureDbosSchema(conn, schema);
+      MigrationManager.ensureMigrationTable(conn, schema);
+      MigrationManager.runDbosMigrations(conn, schema, migrations.subList(0, 125));
+      assertEquals(125, getVersion(conn));
+      if (fourthParameter.equals("message_id")) {
+        recreateSendMessageAsOtherSdksDo(conn);
+      }
+      assertEquals(fourthParameter, sendMessageFourthParameter(conn));
+      assertColumnAbsent(conn, "workflow_status", "suspend_key");
+      assertIndexAbsent(conn, "idx_workflow_status_suspended");
+      assertIndexAbsent(conn, "idx_workflow_status_deadline_v2");
+      assertIndexExists(conn, "idx_workflow_status_deadline");
+    }
+
+    MigrationManager.runMigrations(dbosConfig);
+
+    try (var conn = dataSource.getConnection()) {
+      assertEquals(130, migrations.size());
+      assertEquals(migrations.size(), getVersion(conn));
+      assertSuspendModeSchema(conn);
+    }
+
+    // Re-execute the DDL as well as the runner's normal already-current path.
+    try (var conn = dataSource.getConnection();
+        var stmt = conn.createStatement()) {
+      for (var migration : migrations.subList(125, 130)) {
+        stmt.execute(migration);
+      }
+      assertSuspendModeSchema(conn);
+    }
+    assertDoesNotThrow(() -> MigrationManager.runMigrations(dbosConfig));
+  }
+
+  /**
+   * After migration 130, a send wakes its destination if it is parked SUSPENDED waiting on the
+   * send's topic, and touches no other row.
+   */
+  @Test
+  void testMigration130_SendMessageWakesASuspendedReceiver() throws Exception {
+    MigrationManager.runMigrations(pgContainer.dbosConfig());
+
+    var schema = Constants.DB_SCHEMA;
+    var parked = 9_000_000_000_000L;
+    try (var conn = dataSource.getConnection()) {
+      // {workflow, status, suspend_key}
+      String[][] rows = {
+        {"waiting", "SUSPENDED", "topic"},
+        {"waiting-null-topic", "SUSPENDED", "__null__topic__"},
+        {"other-topic", "SUSPENDED", "elsewhere"},
+        {"not-suspended", "PENDING", "topic"},
+        {"not-addressed", "SUSPENDED", "topic"},
+        {"no-key", "SUSPENDED", null},
+      };
+      var insert =
+          "INSERT INTO \"%s\".workflow_status (workflow_uuid, status, name, suspend_key, delay_until_epoch_ms) VALUES (?, ?, 'wf', ?, ?)"
+              .formatted(schema);
+      try (var ps = conn.prepareStatement(insert)) {
+        for (var row : rows) {
+          ps.setString(1, row[0]);
+          ps.setString(2, row[1]);
+          ps.setString(3, row[2]);
+          ps.setLong(4, parked);
+          ps.addBatch();
+        }
+        ps.executeBatch();
+      }
+
+      sendMessage(conn, "waiting", "topic");
+      sendMessage(conn, "waiting-null-topic", null);
+      sendMessage(conn, "other-topic", "topic");
+      sendMessage(conn, "not-suspended", "topic");
+      sendMessage(conn, "no-key", "topic");
+
+      assertEquals(0L, delayUntil(conn, "waiting"));
+      assertEquals(0L, delayUntil(conn, "waiting-null-topic"));
+      assertEquals(parked, delayUntil(conn, "other-topic"));
+      assertEquals(parked, delayUntil(conn, "not-suspended"));
+      assertEquals(parked, delayUntil(conn, "not-addressed"));
+      assertEquals(parked, delayUntil(conn, "no-key"));
+
+      // Each send still delivered its message, whether or not it woke anything.
+      try (var stmt = conn.createStatement();
+          var rs =
+              stmt.executeQuery("SELECT count(*) FROM \"%s\".notifications".formatted(schema))) {
+        assertTrue(rs.next());
+        assertEquals(5, rs.getInt(1));
+      }
+
+      // A send to a workflow that does not exist still fails as it did before.
+      var e =
+          assertThrows(java.sql.SQLException.class, () -> sendMessage(conn, "missing", "topic"));
+      assertTrue(
+          e.getMessage().contains("DBOS non-existent workflow"),
+          "Expected the non-existent workflow error, got: " + e.getMessage());
+    }
+  }
+
+  private static void sendMessage(Connection conn, String destination, String topic)
+      throws Exception {
+    var sql = "SELECT \"%s\".send_message(?, ?::json, ?)".formatted(Constants.DB_SCHEMA);
+    try (var ps = conn.prepareStatement(sql)) {
+      ps.setString(1, destination);
+      ps.setString(2, "\"hello\"");
+      ps.setString(3, topic);
+      ps.execute();
+    }
+  }
+
+  private static long delayUntil(Connection conn, String workflowId) throws Exception {
+    var sql =
+        "SELECT delay_until_epoch_ms FROM \"%s\".workflow_status WHERE workflow_uuid = ?"
+            .formatted(Constants.DB_SCHEMA);
+    try (var ps = conn.prepareStatement(sql)) {
+      ps.setString(1, workflowId);
+      try (var rs = ps.executeQuery()) {
+        assertTrue(rs.next());
+        return rs.getLong(1);
+      }
+    }
+  }
+
+  /** What migrations 126 to 130 leave behind. */
+  static void assertSuspendModeSchema(Connection conn) throws Exception {
+    assertColumnExists(conn, "workflow_status", "suspend_key");
+    assertIndexExists(conn, "idx_workflow_status_suspended");
+    assertIndexExists(conn, "idx_workflow_status_deadline_v2");
+    assertIndexAbsent(conn, "idx_workflow_status_deadline");
+    assertFunctionExists(conn.getMetaData(), "send_message");
+    assertEquals("message_id", sendMessageFourthParameter(conn));
+    if (!PgContainer.USE_COCKROACH_DB) {
+      assertEquals(
+          List.of("search_path=pg_catalog, pg_temp"),
+          sendMessageConfig(conn),
+          "send_message keeps the search_path migration 20 gave it");
+    }
+  }
+
+  /** The name of send_message's fourth parameter, the one the SDKs disagreed on before 130. */
+  private static String sendMessageFourthParameter(Connection conn) throws Exception {
+    var sql =
+        "SELECT p.proargnames FROM pg_proc p JOIN pg_namespace n ON p.pronamespace = n.oid"
+            + " WHERE n.nspname = ? AND p.proname = 'send_message'";
+    try (var ps = conn.prepareStatement(sql)) {
+      ps.setString(1, Constants.DB_SCHEMA);
+      try (var rs = ps.executeQuery()) {
+        assertTrue(rs.next(), "send_message should exist");
+        var names = (String[]) rs.getArray(1).getArray();
+        assertFalse(rs.next(), "send_message should have one overload");
+        assertEquals(4, names.length);
+        return names[3];
+      }
+    }
+  }
+
+  private static List<String> sendMessageConfig(Connection conn) throws Exception {
+    var sql =
+        "SELECT p.proconfig FROM pg_proc p JOIN pg_namespace n ON p.pronamespace = n.oid"
+            + " WHERE n.nspname = ? AND p.proname = 'send_message'";
+    try (var ps = conn.prepareStatement(sql)) {
+      ps.setString(1, Constants.DB_SCHEMA);
+      try (var rs = ps.executeQuery()) {
+        assertTrue(rs.next(), "send_message should exist");
+        var config = rs.getArray(1);
+        return config == null ? List.of() : List.of((String[]) config.getArray());
+      }
+    }
+  }
+
+  /**
+   * Recreates send_message the way every SDK but Java creates it in migration 14 and secures it in
+   * migration 20: the same function, with its fourth parameter named message_id.
+   */
+  private static void recreateSendMessageAsOtherSdksDo(Connection conn) throws Exception {
+    var schema = Constants.DB_SCHEMA;
+    var sql =
+        """
+        DROP FUNCTION "%1$s".send_message(TEXT, JSON, TEXT, TEXT);
+
+        CREATE FUNCTION "%1$s".send_message(
+            destination_id TEXT,
+            message JSON,
+            topic TEXT DEFAULT NULL,
+            message_id TEXT DEFAULT NULL
+        ) RETURNS VOID AS $$
+        DECLARE
+            v_topic TEXT := COALESCE(topic, '__null__topic__');
+            v_message_id TEXT := COALESCE(message_id, gen_random_uuid()::TEXT);
+        BEGIN
+            INSERT INTO "%1$s".notifications (
+                destination_uuid, topic, message, message_uuid, serialization
+            ) VALUES (
+                destination_id, v_topic, message, v_message_id, 'portable_json'
+            )
+            ON CONFLICT (message_uuid) DO NOTHING;
+        EXCEPTION
+            WHEN foreign_key_violation THEN
+                RAISE EXCEPTION 'DBOS non-existent workflow'
+                   USING DETAIL = format('Destination workflow %%s does not exist', destination_id),
+                        ERRCODE = 'foreign_key_violation';
+        END;
+        $$ LANGUAGE plpgsql;
+        """
+            .formatted(schema);
+    if (!PgContainer.USE_COCKROACH_DB) {
+      sql +=
+          """
+
+          ALTER FUNCTION "%1$s".send_message(
+              TEXT, JSON, TEXT, TEXT
+          ) SET search_path = pg_catalog, pg_temp;
+          """
+              .formatted(schema);
+    }
+    try (var stmt = conn.createStatement()) {
+      stmt.execute(sql);
+    }
+  }
+
   // application_name is the one non-key column: stored in the index, after the key columns.
   private static void assertIncludesApplicationName(Connection conn, String indexName)
       throws Exception {