4258 bug panic on live select as non root user (#4330)

core/src/ctx/context.rs

@@ -248,8 +248,15 @@ impl<'a> Context<'a> {
 		self
 	}
 
+	pub fn get_transaction(&self) -> Option<&Transaction> {
+		self.transaction.as_ref()
+	}
+
 	pub(crate) fn tx_lock(&self) -> MutexLockFuture<'_, kvs::Transaction> {
-		self.transaction.as_ref().map(|txn| txn.lock()).unwrap_or_else(|| unreachable!())
+		self.transaction
+			.as_ref()
+			.map(|txn| txn.lock())
+			.unwrap_or_else(|| unreachable!("The context was not associated with a transaction"))
 	}
 
 	/// Get the timeout for this operation, if any. This is useful for

core/src/doc/lives.rs

@@ -42,7 +42,8 @@ impl<'a> Document<'a> {
 			// Loop through all index statements
 			let lq_stms = self.lv(ctx, opt).await?;
 			let borrows = lq_stms.iter().collect::<Vec<_>>();
-			self.check_lqs_and_send_notifications(stk, opt, stm, borrows.as_slice(), chn).await?;
+			self.check_lqs_and_send_notifications(stk, ctx, opt, stm, borrows.as_slice(), chn)
+				.await?;
 		}
 		// Carry on
 		Ok(())
@@ -104,6 +105,7 @@ impl<'a> Document<'a> {
 	pub(crate) async fn check_lqs_and_send_notifications(
 		&self,
 		stk: &mut Stk,
+		ctx: &Context<'_>,
 		opt: &Options,
 		stm: &Statement<'_>,
 		live_statements: &[&LiveStatement],
@@ -155,7 +157,11 @@ impl<'a> Document<'a> {
 			// use for processing this LIVE query statement.
 			// This ensures that we are using the session
 			// of the user who created the LIVE query.
-			let mut lqctx = Context::background();
+			let lqctx = Context::background();
+			let mut lqctx =
+				lqctx.set_transaction(ctx.get_transaction().cloned().unwrap_or_else(|| {
+					unreachable!("Expected transaction to be available in parent context")
+				}));
 			lqctx.add_value("access", sess.pick(AC.as_ref()));
 			lqctx.add_value("auth", sess.pick(RD.as_ref()));
 			lqctx.add_value("token", sess.pick(TK.as_ref()));

core/src/kvs/ds.rs

@@ -862,9 +862,10 @@ impl Datastore {
 	pub async fn process_lq_notifications(
 		&self,
 		stk: &mut Stk,
+		ctx: &Context<'_>,
 		opt: &Options,
 	) -> Result<(), Error> {
-		process_lq_notifications(self, stk, opt).await
+		process_lq_notifications(self, ctx, stk, opt).await
 	}
 
 	/// Add and kill live queries being track on the datastore
@@ -1227,6 +1228,8 @@ impl Datastore {
 		let ctx = vars.attach(ctx)?;
 		// Start a new transaction
 		let txn = self.transaction(val.writeable().into(), Optimistic).await?.enclose();
+		// Set the context transaction
+		let ctx = ctx.set_transaction(txn.clone());
 		// Compute the value
 		let res = stack.enter(|stk| val.compute(stk, &ctx, &opt, None)).finish().await;
 		// Store any data

core/src/kvs/lq_v2_doc.rs

@@ -197,11 +197,6 @@ mod test_check_lqs_and_send_notifications {
 	use std::collections::BTreeMap;
 	use std::sync::Arc;
 
-	use channel::Sender;
-	use futures::executor::block_on;
-	use once_cell::sync::Lazy;
-	use reblessive::TreeStack;
-
 	use crate::cf::TableMutation;
 	use crate::ctx::Context;
 	use crate::dbs::fuzzy_eq::FuzzyEq;
@@ -209,10 +204,16 @@ mod test_check_lqs_and_send_notifications {
 	use crate::fflags::FFLAGS;
 	use crate::iam::{Auth, Role};
 	use crate::kvs::lq_v2_doc::construct_document;
-	use crate::kvs::Datastore;
+	use crate::kvs::LockType::Optimistic;
+	use crate::kvs::{Datastore, TransactionType};
 	use crate::sql::paths::{OBJ_PATH_ACCESS, OBJ_PATH_AUTH, OBJ_PATH_TOKEN};
 	use crate::sql::statements::{CreateStatement, DeleteStatement, LiveStatement};
 	use crate::sql::{Fields, Object, Strand, Table, Thing, Uuid, Value, Values};
+	use channel::Sender;
+	use futures::executor::block_on;
+	use once_cell::sync::Lazy;
+	use reblessive::TreeStack;
+	use TransactionType::Write;
 
 	static SETUP: Lazy<Arc<TestSuite>> = Lazy::new(|| Arc::new(block_on(setup_test_suite_init())));
 
@@ -275,10 +276,19 @@ mod test_check_lqs_and_send_notifications {
 		// Perform "live query" on the constructed doc that we are checking
 		let live_statement = a_live_query_statement();
 		let executed_statement = a_create_statement();
+		let mut tx = Datastore::new("memory")
+			.await
+			.unwrap()
+			.transaction(Write, Optimistic)
+			.await
+			.unwrap()
+			.enclose();
+		let ctx = Context::background().set_transaction(tx.clone());
 		let mut stack = TreeStack::new();
 		stack.enter(|stk| async {
 			doc.check_lqs_and_send_notifications(
 				stk,
+				&ctx,
 				&opt,
 				&Statement::Create(&executed_statement),
 				&[&live_statement],
@@ -287,6 +297,7 @@ mod test_check_lqs_and_send_notifications {
 			.await
 			.unwrap();
 		});
+		tx.lock().await.commit().await.unwrap();
 
 		// THEN:
 		let notification = receiver.try_recv().expect("There should be a notification");
@@ -323,10 +334,19 @@ mod test_check_lqs_and_send_notifications {
 		// Perform "live query" on the constructed doc that we are checking
 		let live_statement = a_live_query_statement();
 		let executed_statement = a_delete_statement();
+		let mut tx = Datastore::new("memory")
+			.await
+			.unwrap()
+			.transaction(Write, Optimistic)
+			.await
+			.unwrap()
+			.enclose();
+		let ctx = Context::background().set_transaction(tx.clone());
 		let mut stack = TreeStack::new();
 		stack.enter(|stk| async {
 			doc.check_lqs_and_send_notifications(
 				stk,
+				&ctx,
 				&opt,
 				&Statement::Delete(&executed_statement),
 				&[&live_statement],
@@ -335,6 +355,7 @@ mod test_check_lqs_and_send_notifications {
 			.await
 			.unwrap();
 		});
+		tx.lock().await.commit().await.unwrap();
 
 		// THEN:
 		let notification = receiver.try_recv().expect("There should be a notification");

core/src/kvs/lq_v2_fut.rs

@@ -1,5 +1,6 @@
 use crate::cf;
 use crate::cf::ChangeSet;
+use crate::ctx::Context;
 use crate::dbs::{Options, Statement};
 use crate::err::Error;
 use crate::fflags::FFLAGS;
@@ -19,6 +20,7 @@ use tokio::sync::RwLock;
 /// Poll change feeds for live query notifications
 pub async fn process_lq_notifications(
 	ds: &Datastore,
+	ctx: &Context<'_>,
 	stk: &mut Stk,
 	opt: &Options,
 ) -> Result<(), Error> {
@@ -68,7 +70,7 @@ pub async fn process_lq_notifications(
 				.join("\n")
 		);
 		for change_set in change_sets {
-			process_change_set_for_notifications(ds, stk, opt, change_set, &lq_pairs).await?;
+			process_change_set_for_notifications(ds, ctx, stk, opt, change_set, &lq_pairs).await?;
 		}
 	}
 	trace!("Finished process lq successfully");
@@ -131,6 +133,7 @@ async fn populate_relevant_changesets(
 
 async fn process_change_set_for_notifications(
 	ds: &Datastore,
+	ctx: &Context<'_>,
 	stk: &mut Stk,
 	opt: &Options,
 	change_set: ChangeSet,
@@ -165,6 +168,7 @@ async fn process_change_set_for_notifications(
 							channel::bounded(notification_capacity);
 						doc.check_lqs_and_send_notifications(
 							stk,
+							ctx,
 							opt,
 							&Statement::Live(&lq_value.stm),
 							[&lq_value.stm].as_slice(),

lib/src/api/engine/tasks.rs

@@ -16,6 +16,8 @@ use crate::options::EngineOptions;
 use crate::engine::IntervalStream;
 #[cfg(not(target_arch = "wasm32"))]
 use crate::Error as RootError;
+use surrealdb_core::ctx::Context;
+use surrealdb_core::kvs::{LockType, TransactionType};
 #[cfg(not(target_arch = "wasm32"))]
 use tokio::spawn as spawn_future;
 use tokio::sync::oneshot;
@@ -159,8 +161,10 @@ fn live_query_change_feed(
 					// ticker will never return None;
 					let i = v.unwrap();
 					trace!("Live query agent tick: {:?}", i);
+					let tx = dbs.transaction(TransactionType::Write, LockType::Optimistic).await.unwrap();
+					let ctx = Context::background().set_transaction(tx.enclose());
 					if let Err(e) =
-						stack.enter(|stk| dbs.process_lq_notifications(stk, &opt)).finish().await
+						stack.enter(|stk| dbs.process_lq_notifications(stk, &ctx, &opt)).finish().await
 					{
 						error!("Error running node agent tick: {}", e);
 						break;

lib/tests/live.rs

@@ -6,6 +6,7 @@ use surrealdb::dbs::Session;
 use surrealdb::err::Error;
 use surrealdb::fflags::FFLAGS;
 use surrealdb::sql::Value;
+use surrealdb_core::dbs::Options;
 
 #[tokio::test]
 async fn live_query_fails_if_no_change_feed() -> Result<(), Error> {
@@ -87,8 +88,9 @@ async fn live_query_sends_registered_lq_details() -> Result<(), Error> {
 	let result = res.remove(0);
 	assert!(result.result.is_ok());
 
-	let def = Default::default();
-	stack.enter(|stk| dbs.process_lq_notifications(stk, &def)).finish().await?;
+	let ctx = Default::default();
+	let opt = Options::default();
+	stack.enter(|stk| dbs.process_lq_notifications(stk, &ctx, &opt)).finish().await?;
 
 	let notifications_chan = dbs.notifications().unwrap();
 
@@ -148,8 +150,9 @@ async fn live_query_does_not_drop_notifications() -> Result<(), Error> {
 			assert!(result.result.is_ok());
 		}
 		// Process the notifications
-		let def = Default::default();
-		stack.enter(|stk| dbs.process_lq_notifications(stk, &def)).finish().await?;
+		let ctx = Default::default();
+		let opt = Options::default();
+		stack.enter(|stk| dbs.process_lq_notifications(stk, &ctx, &opt)).finish().await?;
 
 		let notifications_chan = dbs.notifications().unwrap();