feat(rivetkit): streamline workflow host APIs

.github/workflows/publish.yaml

@@ -399,6 +399,8 @@ jobs:
           node-version: "22"
           registry-url: "https://registry.npmjs.org"
           cache: pnpm
+      - name: Install OIDC-capable npm
+        run: npm install --global npm@11.16.0
       - run: pnpm install --frozen-lockfile
       - uses: ./.github/actions/docker-setup
         with:
@@ -568,7 +570,6 @@ jobs:
 
       - name: Publish npm packages
         env:
-          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
           SKIP_WASM_BUILD: "1"
         run: |
           pnpm --filter=publish exec tsx src/ci/bin.ts publish-npm \

rivetkit-rust/engine/artifacts/errors/queue.message_identity_mismatch.json

@@ -0,0 +1,5 @@
+{
+  "code": "message_identity_mismatch",
+  "group": "queue",
+  "message": "Queue message identity does not match"
+}

rivetkit-rust/packages/actor-persist/src/versioned.rs

@@ -491,3 +491,47 @@ impl OwnedVersionedData for LastPushedAlarm {
 		Vec::<fn(Self) -> Result<Self>>::new()
 	}
 }
+
+/// Logical run-handler deadline stored separately from the shared physical
+/// alarm. This uses its own versioned type so its persisted meaning cannot be
+/// conflated with `LastPushedAlarm` even though both currently encode an
+/// optional millisecond timestamp.
+pub enum RunWakeAt {
+	V1(Option<i64>),
+}
+
+impl OwnedVersionedData for RunWakeAt {
+	type Latest = Option<i64>;
+
+	fn wrap_latest(latest: Self::Latest) -> Self {
+		Self::V1(latest)
+	}
+
+	fn unwrap_latest(self) -> Result<Self::Latest> {
+		match self {
+			Self::V1(data) => Ok(data),
+		}
+	}
+
+	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
+		match version {
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			_ => bail!("invalid run wake deadline version: {version}"),
+		}
+	}
+
+	fn serialize_version(self, version: u16) -> Result<Vec<u8>> {
+		match (self, version) {
+			(Self::V1(data), 1) => serde_bare::to_vec(&data).map_err(Into::into),
+			(_, version) => bail!("unexpected run wake deadline version: {version}"),
+		}
+	}
+
+	fn deserialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		Vec::<fn(Self) -> Result<Self>>::new()
+	}
+
+	fn serialize_converters() -> Vec<impl Fn(Self) -> Result<Self>> {
+		Vec::<fn(Self) -> Result<Self>>::new()
+	}
+}

rivetkit-rust/packages/actor-persist/tests/versioned.rs

@@ -27,3 +27,15 @@ fn actor_decodes_legacy_raw_v4_when_current_v4_accepts_bytes() {
 
 	assert_eq!(decoded.scheduled_events[0].args, Some(vec![0x01, 0x99]));
 }
+
+#[test]
+fn run_wake_deadline_round_trips_with_embedded_version() {
+	let encoded = versioned::RunWakeAt::wrap_latest(Some(1_725_000_000_123))
+		.serialize_with_embedded_version(1)
+		.expect("encode run wake deadline");
+	assert_eq!(&encoded[..2], &[1, 0]);
+
+	let decoded = versioned::RunWakeAt::deserialize_with_embedded_version(&encoded)
+		.expect("decode run wake deadline");
+	assert_eq!(decoded, Some(1_725_000_000_123));
+}

rivetkit-rust/packages/rivetkit-core/schemas/testing/workflow-fixture/v1.bare

@@ -0,0 +1,83 @@
+# Test-only logical actor SQLite fixture.
+#
+# The Rust definitions in `src/testing/workflow_fixture.rs` intentionally mirror
+# this schema. The outer payload is encoded with vbare's embedded version header.
+
+type FixtureMetadata struct {
+	fixtureName: str
+	sourceRivetkitVersion: str
+	sourceWorkflowVersion: str
+	sourceRevision: str
+	actorId: str
+	registryKey: str
+	internalSchemaVersion: i64
+	fakeClockSeed: u64
+	generatedIdSeed: u64
+}
+
+type RuntimeRow struct {
+	lastPushedAlarm: optional<i64>
+	inspectorToken: optional<str>
+	queueNextId: i64
+}
+
+type MetaRow struct {
+	key: str
+	value: data
+}
+
+type ActorRow struct {
+	hasInitialized: i64
+	input: optional<data>
+}
+
+type WorkflowRow struct {
+	key: data
+	value: data
+}
+
+type QueueRow struct {
+	id: i64
+	name: str
+	body: data
+	createdAt: i64
+}
+
+type ScheduleEventRow struct {
+	eventId: str
+	triggerAt: i64
+	action: str
+	args: optional<data>
+	kind: i64
+	cronExpression: optional<str>
+	timezone: optional<str>
+	intervalMs: optional<i64>
+	lastStartedAt: optional<i64>
+	maxHistory: i64
+}
+
+type ScheduleHistoryRow struct {
+	id: i64
+	scheduleId: str
+	action: str
+	scheduledAt: i64
+	firedAt: i64
+	finishedAt: optional<i64>
+	result: i64
+	errorGroup: optional<str>
+	errorCode: optional<str>
+	errorMessage: optional<str>
+	errorMetadata: optional<data>
+}
+
+type WorkflowFixture struct {
+	metadata: FixtureMetadata
+	metaRows: list<MetaRow>
+	runtime: optional<RuntimeRow>
+	actor: optional<ActorRow>
+	actorState: optional<data>
+	workflowRows: list<WorkflowRow>
+	queueRows: list<QueueRow>
+	scheduleEvents: list<ScheduleEventRow>
+	scheduleHistory: list<ScheduleHistoryRow>
+}

rivetkit-rust/packages/rivetkit-core/src/actor/context.rs

@@ -75,6 +75,7 @@
 	pub(super) current_state: RwLock<Vec<u8>>,
 	pub(super) persisted: RwLock<PersistedActor>,
 	pub(super) last_pushed_alarm: RwLock<Option<i64>>,
+	pub(super) run_wake_at: RwLock<Option<i64>>,
 	pub(super) state_save_interval: Duration,
 	pub(super) state_dirty: AtomicBool,
 	pub(super) state_revision: AtomicU64,
@@ -88,7 +89,7 @@
 	pub(super) last_save_at: Mutex<Option<crate::time::Instant>>,
 	pub(super) pending_save: Mutex<Option<PendingSave>>,
 	pub(super) tracked_persist: Mutex<Option<JoinHandle<()>>>,
-	pub(super) save_guard: AsyncMutex<()>,
+	pub(super) save_guard: Arc<AsyncMutex<()>>,
 	pub(super) in_flight_state_writes: AtomicUsize,
 	pub(super) state_write_completion: Notify,
 	pub(super) on_state_change_in_flight: AtomicUsize,
@@ -112,6 +113,7 @@
 	// being moved out of the lock.
 	pub(super) schedule_pending_alarm_writes: Mutex<Vec<oneshot::Receiver<()>>>,
 	pub(super) schedule_local_alarm_epoch: AtomicU64,
+	pub(super) schedule_alarm_push_epoch: AtomicU64,
 	pub(super) schedule_alarm_dispatch_enabled: AtomicBool,
 	pub(super) schedule_dirty_since_push: AtomicBool,
 	pub(super) schedule_mutation_lock: AsyncMutex<()>,
@@ -297,6 +299,7 @@
 			current_state: RwLock::new(Vec::new()),
 			persisted: RwLock::new(PersistedActor::default()),
 			last_pushed_alarm: RwLock::new(None),
+			run_wake_at: RwLock::new(None),
 			state_save_interval,
 			state_dirty: AtomicBool::new(false),
 			state_revision: AtomicU64::new(0),
@@ -309,7 +312,7 @@
 			last_save_at: Mutex::new(None),
 			pending_save: Mutex::new(None),
 			tracked_persist: Mutex::new(None),
-			save_guard: AsyncMutex::new(()),
+			save_guard: Arc::new(AsyncMutex::new(())),
 			in_flight_state_writes: AtomicUsize::new(0),
 			state_write_completion: Notify::new(),
 			on_state_change_in_flight: AtomicUsize::new(0),
@@ -326,6 +329,7 @@
 			schedule_local_alarm_task: Mutex::new(None),
 			schedule_pending_alarm_writes: Mutex::new(Vec::new()),
 			schedule_local_alarm_epoch: AtomicU64::new(0),
+			schedule_alarm_push_epoch: AtomicU64::new(0),
 			schedule_alarm_dispatch_enabled: AtomicBool::new(true),
 			// A fresh actor context has no in-process record of a successful
 			// envoy alarm push yet, so the first sync must always push.
@@ -1123,7 +1127,7 @@
 		*self.0.hibernated_connection_liveness_override.write() = Some(pairs.into_iter().collect());
 	}
 
-	fn prepare_state_deltas(
+	pub(super) fn prepare_state_deltas(
 		&self,
 		deltas: Vec<StateDelta>,
 	) -> Result<(Vec<StateDelta>, PendingHibernationChanges)> {
@@ -1403,7 +1407,7 @@
 		self.reset_sleep_timer();
 	}
 
 	pub(crate) fn sleep_config(&self) -> ActorConfig {
 		self.sleep_state_config()
 	}
 

rivetkit-rust/packages/rivetkit-core/src/actor/internal_storage/mod.rs

@@ -2,12 +2,16 @@ use std::collections::{BTreeSet, HashMap};
 use std::io::Cursor;
 
 use anyhow::{Context, Result, bail};
+use rivetkit_actor_persist::versioned as persist_versioned;
 
 use crate::actor::connection::{
 	PersistedConnection, PersistedSubscription, encode_persisted_connection,
 };
 use crate::actor::keys::make_workflow_key;
 use crate::actor::messages::WorkflowKvWrite;
+use crate::actor::persist::{
+	decode_latest_with_embedded_version, encode_latest_with_embedded_version,
+};
 use crate::actor::queue::{PersistedQueueMessage, QueueMetadata};
 use crate::actor::state::PersistedActor;
 use crate::error::KvRuntimeError;
@@ -28,6 +32,8 @@ pub(crate) const USER_KV_BATCH_GET_MAX_KEYS: usize = 128;
 const QUEUE_METADATA_PAGE_ROWS: usize = 128;
 const QUEUE_MESSAGE_IDS_PER_QUERY: usize = 128;
 const WORKFLOW_KV_VALUE_LIMIT: usize = 256 * 1024;
+pub(crate) const RUN_WAKE_AT_META_KEY: &str = "run_wake_at";
+const RUN_WAKE_AT_VERSION: u16 = 1;
 
 /// Depot rejects SQLite commits that dirty more than `MAX_COMMIT_RAW_DIRTY_BYTES`
 /// (320 pages * 4 KiB = 1.3 MiB) in `engine/packages/depot/src/conveyer/constants.rs`.
@@ -72,6 +78,7 @@ where
 pub(crate) struct InternalActorSnapshot {
 	pub actor: PersistedActor,
 	pub last_pushed_alarm: Option<i64>,
+	pub run_wake_at: Option<i64>,
 }
 
 pub(crate) async fn load_actor_snapshot(db: &SqliteDb) -> Result<Option<InternalActorSnapshot>> {
@@ -87,6 +94,7 @@ pub(crate) async fn load_actor_snapshot(db: &SqliteDb) -> Result<Option<Internal
 	let input = read_optional_blob(row, 1, "input")?;
 	let state = read_blob(row, 2, "state")?;
 	let last_pushed_alarm = load_last_pushed_alarm(db).await?;
+	let run_wake_at = load_run_wake_at(db).await?;
 
 	Ok(Some(InternalActorSnapshot {
 		actor: PersistedActor {
@@ -96,6 +104,7 @@ pub(crate) async fn load_actor_snapshot(db: &SqliteDb) -> Result<Option<Internal
 			scheduled_events: Vec::new(),
 		},
 		last_pushed_alarm,
+		run_wake_at,
 	}))
 }
 
@@ -209,7 +218,7 @@ pub(crate) async fn persist_actor_core_connections_and_workflow(
 	Ok(())
 }
 
-fn build_actor_core_and_connection_statements(
+pub(crate) fn build_actor_core_and_connection_statements(
 	actor: Option<&PersistedActor>,
 	connections: &[PersistedConnection],
 	removed_connections: &[String],
@@ -477,6 +486,21 @@ pub(crate) async fn load_queue_messages(db: &SqliteDb) -> Result<Vec<QueueMessag
 	decode_queue_message_rows(&result.rows)
 }
 
+pub(crate) async fn load_queue_message_name(db: &SqliteDb, id: u64) -> Result<Option<String>> {
+	let id = i64::try_from(id).context("queue message id exceeds sqlite integer range")?;
+	let result = db
+		.query(
+			LOAD_QUEUE_MESSAGE_NAME_SQL,
+			Some(vec![BindParam::Integer(id)]),
+		)
+		.await
+		.context("load internal queue message name")?;
+	let Some(row) = result.rows.first() else {
+		return Ok(None);
+	};
+	Ok(Some(read_text(row, 0, "queue message name")?))
+}
+
 pub(crate) async fn load_queue_messages_matching(
 	db: &SqliteDb,
 	names: Option<&BTreeSet<String>>,
@@ -855,22 +879,33 @@ fn build_workflow_kv_statements(writes: &[WorkflowKvWrite]) -> Result<Vec<Sqlite
 
 fn validate_atomic_workflow_flush(statements: &[SqliteBatchStatement]) -> Result<()> {
 	let row_count = statements.len();
-	let payload_bytes = statements
-		.iter()
-		.flat_map(|statement| statement.params.iter().flatten())
-		.map(bind_param_payload_len)
-		.fold(0usize, usize::saturating_add);
+	let payload_bytes = statement_bind_payload_len(statements);
+	validate_atomic_state_transaction_budget(row_count, payload_bytes)
+}
+
+pub(crate) fn validate_atomic_state_transaction_budget(
+	row_count: usize,
+	payload_bytes: usize,
+) -> Result<()> {
 	if row_count > KV_TX_MAX_ROWS || payload_bytes > KV_TX_MAX_PAYLOAD_BYTES {
 		bail!(
-			"atomic actor state and workflow flush exceeds sqlite transaction budget: {row_count} rows and {payload_bytes} bytes (limits: {} rows and {} bytes)",
+			"atomic SQLite and actor state transaction exceeds transaction budget: {row_count} rows and {payload_bytes} bytes (limits: {} rows and {} bytes)",
 			KV_TX_MAX_ROWS,
 			KV_TX_MAX_PAYLOAD_BYTES
 		);
 	}
 	Ok(())
 }
 
-fn bind_param_payload_len(param: &BindParam) -> usize {
+pub(crate) fn statement_bind_payload_len(statements: &[SqliteBatchStatement]) -> usize {
+	statements
+		.iter()
+		.flat_map(|statement| statement.params.iter().flatten())
+		.map(bind_param_payload_len)
+		.fold(0usize, usize::saturating_add)
+}
+
+pub(crate) fn bind_param_payload_len(param: &BindParam) -> usize {
 	match param {
 		BindParam::Null => 0,
 		BindParam::Integer(_) | BindParam::Float(_) => 8,
@@ -995,6 +1030,42 @@ pub(crate) async fn persist_last_pushed_alarm(db: &SqliteDb, alarm_ts: Option<i6
 	Ok(())
 }
 
+pub(crate) async fn load_run_wake_at(db: &SqliteDb) -> Result<Option<i64>> {
+	let result = db
+		.query(
+			LOAD_RUN_WAKE_AT_SQL,
+			Some(vec![BindParam::Text(RUN_WAKE_AT_META_KEY.to_owned())]),
+		)
+		.await
+		.context("load internal run wake deadline")?;
+	let Some(row) = result.rows.first() else {
+		return Ok(None);
+	};
+	let payload = read_blob(row, 0, "run wake deadline")?;
+	decode_latest_with_embedded_version::<persist_versioned::RunWakeAt>(
+		&payload,
+		"run wake deadline",
+	)
+}
+
+pub(crate) async fn persist_run_wake_at(db: &SqliteDb, wake_at: Option<i64>) -> Result<()> {
+	let payload = encode_latest_with_embedded_version::<persist_versioned::RunWakeAt>(
+		wake_at,
+		RUN_WAKE_AT_VERSION,
+		"run wake deadline",
+	)?;
+	db.execute(
+		UPSERT_RUN_WAKE_AT_SQL,
+		Some(vec![
+			BindParam::Text(RUN_WAKE_AT_META_KEY.to_owned()),
+			BindParam::Blob(payload),
+		]),
+	)
+	.await
+	.context("persist internal run wake deadline")?;
+	Ok(())
+}
+
 pub(crate) async fn load_inspector_token(db: &SqliteDb) -> Result<Option<String>> {
 	let result = db
 		.query(LOAD_INSPECTOR_TOKEN_SQL, None)