fix: surface chat state read failures

This commit is contained in:
2026-09-24 19:37:29 +08:00
parent 412359e84c
commit 8428468712
8 changed files with 110 additions and 88 deletions
+2 -1
View File
@@ -131,7 +131,8 @@ pub(crate) mod test_support {
}, },
); );
}) })
.await; .await
.expect("seed prompt state");
} }
pub(crate) struct TestStores { pub(crate) struct TestStores {
+8 -4
View File
@@ -59,7 +59,8 @@ async fn handle_callback(
}; };
// Lazy expiry: a stale record (past the TTL, not yet swept) is dropped. // Lazy expiry: a stale record (past the TTL, not yet swept) is dropped.
if edit.created_at + ttl_secs <= unix_now() { if edit.created_at + ttl_secs <= unix_now() {
ctx.chat_store let _ = ctx
.chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.edit_message.remove(&prompt_message_id); data.edit_message.remove(&prompt_message_id);
}) })
@@ -80,7 +81,8 @@ async fn handle_callback(
// "do not forward this" answer, and it drops the record so the forward // "do not forward this" answer, and it drops the record so the forward
// can never happen later. // can never happen later.
log::info!("edit-before-forward prompt {prompt_message_id} skipped"); log::info!("edit-before-forward prompt {prompt_message_id} skipped");
ctx.chat_store let _ = ctx
.chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.edit_message.remove(&prompt_message_id); data.edit_message.remove(&prompt_message_id);
}) })
@@ -150,7 +152,8 @@ async fn handle_callback(
.sender .sender
.delete_message(ChatId(chat_id), MessageId(prompt_message_id as i32)) .delete_message(ChatId(chat_id), MessageId(prompt_message_id as i32))
.await; .await;
ctx.chat_store let _ = ctx
.chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.edit_message.remove(&prompt_message_id); data.edit_message.remove(&prompt_message_id);
}) })
@@ -190,7 +193,8 @@ async fn handle_callback(
.await .await
{ {
super::EditOutcome::Applied => { super::EditOutcome::Applied => {
ctx.chat_store let _ = ctx
.chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
if let Some(entry) = data.edit_message.get_mut(&prompt_message_id) { if let Some(entry) = data.edit_message.get_mut(&prompt_message_id) {
entry.template = name.to_string(); entry.template = name.to_string();
+66 -52
View File
@@ -74,6 +74,8 @@ fn parse_arg_remainder(s: String) -> Result<(String,), ParseError> {
/// `x_media::site::caption_from_fields` substitutes. /// `x_media::site::caption_from_fields` substitutes.
const FORMAT_PLACEHOLDERS: [&str; 6] = ["url", "author", "author_url", "title", "content", "tags"]; const FORMAT_PLACEHOLDERS: [&str; 6] = ["url", "author", "author_url", "title", "content", "tags"];
const CHAT_STATE_READ_ERROR: &str = "Couldn't read chat settings; try again.";
/// `/start`'s welcome: what the bot is for, where links work, where to look /// `/start`'s welcome: what the bot is for, where links work, where to look
/// next. The old "Hello!" left a first-time user with nothing. /// next. The old "Hello!" left a first-time user with nothing.
const START_TEXT: &str = "\ const START_TEXT: &str = "\
@@ -294,14 +296,19 @@ pub(crate) async fn execute_command(
} }
Command::SetForwardChannel(channel) => { Command::SetForwardChannel(channel) => {
let result = match set_forward_channel_handler(bot, message, channel).await { let result = match set_forward_channel_handler(bot, message, channel).await {
Ok(channel_id) => { Ok(channel_id) => match ctx
ctx.chat_store .chat_store
.update(message.chat.id.0, |data| { .update(message.chat.id.0, |data| {
data.forward_channel_id = Some(channel_id); data.forward_channel_id = Some(channel_id);
}) })
.await; .await
"Add successfully.".to_string() {
} Ok((_, true)) => "Add successfully.".to_string(),
Ok((_, false)) => {
"Forward channel set only in memory; retry later.".to_string()
}
Err(()) => CHAT_STATE_READ_ERROR.to_string(),
},
Err(SetForwardChannelError::EmptyParameter) => { Err(SetForwardChannelError::EmptyParameter) => {
"Receive empty parameter.\nYou should enter a channel id or username" "Receive empty parameter.\nYou should enter a channel id or username"
.to_string() .to_string()
@@ -323,7 +330,7 @@ pub(crate) async fn execute_command(
} }
Command::RemoveForwardChannel => { Command::RemoveForwardChannel => {
let chat_id = message.chat.id.0; let chat_id = message.chat.id.0;
let (text, _) = ctx let text = match ctx
.chat_store .chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
if data.forward_channel_id.is_some() { if data.forward_channel_id.is_some() {
@@ -333,12 +340,16 @@ pub(crate) async fn execute_command(
"No channel to remove.".to_string() "No channel to remove.".to_string()
} }
}) })
.await; .await
{
Ok((text, _)) => text,
Err(()) => CHAT_STATE_READ_ERROR.to_string(),
};
reply(ctx.sender, message.chat.id.0, message.id, text).await?; reply(ctx.sender, message.chat.id.0, message.id, text).await?;
} }
Command::EditBeforeForward => { Command::EditBeforeForward => {
let chat_id = message.chat.id.0; let chat_id = message.chat.id.0;
let (text, _) = ctx let text = match ctx
.chat_store .chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
if data.forward_channel_id.is_none() { if data.forward_channel_id.is_none() {
@@ -352,7 +363,11 @@ pub(crate) async fn execute_command(
"Enable edit before forward.".to_string() "Enable edit before forward.".to_string()
} }
}) })
.await; .await
{
Ok((text, _)) => text,
Err(()) => CHAT_STATE_READ_ERROR.to_string(),
};
reply(ctx.sender, message.chat.id.0, message.id, text).await?; reply(ctx.sender, message.chat.id.0, message.id, text).await?;
} }
Command::SetTemplate(name) => { Command::SetTemplate(name) => {
@@ -366,15 +381,22 @@ pub(crate) async fn execute_command(
} else if name.is_empty() { } else if name.is_empty() {
"Please provide a name for the template.".to_string() "Please provide a name for the template.".to_string()
} else { } else {
ctx.chat_store match ctx
.chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.template.insert( data.template.insert(
name, name,
html_escape::encode_text(reply_text).into_owned(), html_escape::encode_text(reply_text).into_owned(),
); );
}) })
.await; .await
"Template set.".to_string() {
Ok((_, true)) => "Template set.".to_string(),
Ok((_, false)) => {
"Template set only in memory; retry later.".to_string()
}
Err(()) => CHAT_STATE_READ_ERROR.to_string(),
}
} }
} }
}; };
@@ -393,21 +415,21 @@ pub(crate) async fn execute_command(
.await?; .await?;
return Ok(()); return Ok(());
} }
let (removed, _) = ctx let text = match ctx
.chat_store .chat_store
.update(chat_id, |data| data.template.remove(&name).is_some()) .update(chat_id, |data| data.template.remove(&name).is_some())
.await; .await
let text = if removed { {
format!("Template '{name}' removed.") Ok((true, _)) => format!("Template '{name}' removed."),
} else { Ok((false, _)) => {
// Name the live templates: a typo would otherwise look like a let names = sorted_template_names(&ctx.chat_store.get(chat_id).await);
// successful delete. if names.is_empty() {
let names = sorted_template_names(&ctx.chat_store.get(chat_id).await); format!("No template named '{name}'. None are saved yet.")
if names.is_empty() { } else {
format!("No template named '{name}'. None are saved yet.") format!("No template named '{name}'. Saved: {}", names.join(", "))
} else { }
format!("No template named '{name}'. Saved: {}", names.join(", "))
} }
Err(()) => CHAT_STATE_READ_ERROR.to_string(),
}; };
reply(ctx.sender, chat_id, message.id, text).await?; reply(ctx.sender, chat_id, message.id, text).await?;
} }
@@ -461,23 +483,18 @@ pub(crate) async fn execute_command(
// set a format once could never get back to the default (the // set a format once could never get back to the default (the
// built-in format string is not something a user can retype). // built-in format string is not something a user can retype).
if format == "-" { if format == "-" {
let (_, saved) = ctx let text = match ctx
.chat_store .chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.message_format.remove(site); data.message_format.remove(site);
}) })
.await; .await
reply( {
ctx.sender, Ok((_, true)) => "Format reset to the built-in one.".to_string(),
message.chat.id.0, Ok((_, false)) => "Reset in memory only; retry later.".to_string(),
message.id, Err(()) => CHAT_STATE_READ_ERROR.to_string(),
if saved { };
"Format reset to the built-in one.".to_string() reply(ctx.sender, message.chat.id.0, message.id, text).await?;
} else {
"Reset in memory only: the database write failed, so it will be lost on restart.".to_string()
},
)
.await?;
return Ok(()); return Ok(());
} }
// A typo like {titel} would otherwise be rendered literally into // A typo like {titel} would otherwise be rendered literally into
@@ -500,23 +517,20 @@ pub(crate) async fn execute_command(
.await?; .await?;
return Ok(()); return Ok(());
} }
let (_, saved) = ctx let text = match ctx
.chat_store .chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.message_format.insert(site.to_string(), format); data.message_format.insert(site.to_string(), format);
}) })
.await; .await
reply( {
ctx.sender, Ok((_, true)) => {
message.chat.id.0,
message.id,
if saved {
"Format set. Use /debug <link> to preview the caption.".to_string() "Format set. Use /debug <link> to preview the caption.".to_string()
} else { }
"Set in memory only: the database write failed, so it will be lost on restart. Use /debug <link> to preview the caption.".to_string() Ok((_, false)) => "Format set only in memory; retry later.".to_string(),
}, Err(()) => CHAT_STATE_READ_ERROR.to_string(),
) };
.await?; reply(ctx.sender, message.chat.id.0, message.id, text).await?;
} }
Command::ClearCache(arg) => { Command::ClearCache(arg) => {
let Some(sender_id) = require_admin(ctx, message).await? else { let Some(sender_id) = require_admin(ctx, message).await? else {
+2 -1
View File
@@ -145,7 +145,8 @@ async fn edit_message_handler(
// (not yet swept) is dropped and the reply falls through to the normal // (not yet swept) is dropped and the reply falls through to the normal
// message flow instead of rewriting a caption from a dead prompt. // message flow instead of rewriting a caption from a dead prompt.
if edit.created_at + ctx.config.edit_message_ttl.as_secs() as i64 <= crate::db::unix_now() { if edit.created_at + ctx.config.edit_message_ttl.as_secs() as i64 <= crate::db::unix_now() {
ctx.chat_store let _ = ctx
.chat_store
.update(chat_id, |data| { .update(chat_id, |data| {
data.edit_message.remove(&reply_to_message_id); data.edit_message.remove(&reply_to_message_id);
}) })
+2 -1
View File
@@ -1009,7 +1009,8 @@ mod tests {
data.forward_channel_id = Some(2); data.forward_channel_id = Some(2);
data.edit_before_forward = true; data.edit_before_forward = true;
}) })
.await; .await
.unwrap();
} }
#[tokio::test] #[tokio::test]
+2 -1
View File
@@ -448,7 +448,8 @@ mod tests {
}, },
); );
}) })
.await; .await
.unwrap();
let (stop_tx, stop_rx) = watch::channel(false); let (stop_tx, stop_rx) = watch::channel(false);
let sweep = periodic_sweep(stores.ctx(&sender), stop_rx); let sweep = periodic_sweep(stores.ctx(&sender), stop_rx);
+2 -1
View File
@@ -301,7 +301,8 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message
log_key(&source_url) log_key(&source_url)
); );
let source_url = source_url.clone(); let source_url = source_url.clone();
ctx.chat_store let _ = ctx
.chat_store
.update(chat_id, move |data| { .update(chat_id, move |data| {
data.edit_message.insert( data.edit_message.insert(
prompt_id, prompt_id,
+26 -27
View File
@@ -151,25 +151,17 @@ impl ChatStore {
/// e.g. a second `edit_message` record. The per-chat lock makes the /// e.g. a second `edit_message` record. The per-chat lock makes the
/// cycle atomic. Returns the closure's result plus whether the DB write /// cycle atomic. Returns the closure's result plus whether the DB write
/// landed (see [`Self::set`]); callers that do not care ignore the flag. /// landed (see [`Self::set`]); callers that do not care ignore the flag.
pub async fn update<R: Default>( pub async fn update<R>(
&self, &self,
chat_id: i64, chat_id: i64,
f: impl FnOnce(&mut ChatData) -> R, f: impl FnOnce(&mut ChatData) -> R,
) -> (R, bool) { ) -> Result<(R, bool), ()> {
let lock = self.lock_for(chat_id); let lock = self.lock_for(chat_id);
let _guard = lock.lock().await; let _guard = lock.lock().await;
let mut data = match self.load(chat_id).await { let mut data = self.load(chat_id).await?;
Ok(data) => data,
Err(()) => {
// Do not run the mutation closure on an empty fallback: a
// command could otherwise report success after the failed
// read and the next update could write those defaults back.
return (R::default(), false);
}
};
let r = f(&mut data); let r = f(&mut data);
let saved = self.set(chat_id, &data).await; let saved = self.set(chat_id, &data).await;
(r, saved) Ok((r, saved))
} }
/// Removes edit-before-forward records whose `created_at + ttl` is in the /// Removes edit-before-forward records whose `created_at + ttl` is in the
@@ -276,7 +268,8 @@ mod tests {
}, },
); );
}) })
.await; .await
.unwrap();
})); }));
} }
for h in handles { for h in handles {
@@ -312,7 +305,8 @@ mod tests {
data.edit_message.insert(1, edit_entry(7, now - 3600)); data.edit_message.insert(1, edit_entry(7, now - 3600));
data.edit_message.insert(2, edit_entry(7, now)); data.edit_message.insert(2, edit_entry(7, now));
}) })
.await; .await
.unwrap();
let removed = store.prune_expired(Duration::from_secs(60)).await; let removed = store.prune_expired(Duration::from_secs(60)).await;
@@ -339,7 +333,8 @@ mod tests {
data.forward_channel_id = Some(-100); data.forward_channel_id = Some(-100);
data.message_format.insert("twitter".into(), "{url}".into()); data.message_format.insert("twitter".into(), "{url}".into());
}) })
.await; .await
.unwrap();
assert!(store.cache.lock().contains_key(&9)); assert!(store.cache.lock().contains_key(&9));
let removed = store.prune_expired(Duration::from_secs(60)).await; let removed = store.prune_expired(Duration::from_secs(60)).await;
@@ -368,7 +363,8 @@ mod tests {
.update(10, |data| { .update(10, |data| {
data.edit_message.insert(1, edit_entry(10, unix_now())); data.edit_message.insert(1, edit_entry(10, unix_now()));
}) })
.await; .await
.unwrap();
store.prune_expired(Duration::from_secs(3600)).await; store.prune_expired(Duration::from_secs(3600)).await;
@@ -390,7 +386,8 @@ mod tests {
data.template.insert("keep".into(), "[]".into()); data.template.insert("keep".into(), "[]".into());
data.edit_message.insert(1, edit_entry(8, 0)); data.edit_message.insert(1, edit_entry(8, 0));
}) })
.await; .await
.unwrap();
let removed = store.prune_expired(Duration::from_secs(60)).await; let removed = store.prune_expired(Duration::from_secs(60)).await;
@@ -446,30 +443,32 @@ mod tests {
let path = dir.path().join("update.db"); let path = dir.path().join("update.db");
let pool = crate::db::open_store(path.to_str().unwrap()).unwrap(); let pool = crate::db::open_store(path.to_str().unwrap()).unwrap();
let raw = rusqlite::Connection::open(&path).unwrap(); let raw = rusqlite::Connection::open(&path).unwrap();
let real = ChatData { let stored = "{\"forward_channel_id\":";
forward_channel_id: Some(42),
..ChatData::default()
};
raw.execute( raw.execute(
"INSERT INTO chat_state (chat_id, payload) VALUES ('7', ?1)", "INSERT INTO chat_state (chat_id, payload) VALUES ('7', ?1)",
rusqlite::params![serde_json::to_string(&real).unwrap()], rusqlite::params![stored],
) )
.unwrap(); .unwrap();
raw.execute_batch("DROP TABLE chat_state").unwrap();
let store = ChatStore::new(pool); let store = ChatStore::new(pool);
let mut called = false; let mut called = false;
let (result, saved) = store let result = store
.update(7, |data| { .update(7, |data| {
called = true; called = true;
data.message_format.insert("twitter".into(), "{url}".into()); data.message_format.insert("twitter".into(), "{url}".into());
}) })
.await; .await;
assert_eq!(result, ()); assert!(result.is_err());
assert!(!saved);
assert!(!called, "a failed load must not run a destructive mutation"); assert!(!called, "a failed load must not run a destructive mutation");
assert!(!store.cache.lock().contains_key(&7)); assert!(!store.cache.lock().contains_key(&7));
let payload: String = raw
.query_row(
"SELECT payload FROM chat_state WHERE chat_id='7'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(payload, stored, "the original row must remain unchanged");
} }
#[tokio::test] #[tokio::test]