mirror of
https://github.com/chatmail/core.git
synced 2026-05-10 02:16:30 +03:00
Compare commits
4 Commits
iequidoo/s
...
link2xt/re
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4f919cee7d | ||
|
|
f93e7761da | ||
|
|
6fb2f27831 | ||
|
|
58a09df49a |
@@ -1091,6 +1091,7 @@ def test_rename_group(acfactory):
|
||||
bob.wait_for_event(EventType.CHATLIST_ITEM_CHANGED)
|
||||
|
||||
for name in ["Baz", "Foo bar", "Xyzzy"]:
|
||||
time.sleep(1)
|
||||
alice_group.set_name(name)
|
||||
bob.wait_for_event(EventType.CHATLIST_ITEM_CHANGED)
|
||||
bob.wait_for_event(EventType.CHATLIST_ITEM_CHANGED)
|
||||
|
||||
62
src/chat.rs
62
src/chat.rs
@@ -49,8 +49,8 @@ use crate::stock_str;
|
||||
use crate::sync::{self, Sync::*, SyncData};
|
||||
use crate::tools::{
|
||||
IsNoneOrEmpty, SystemTime, buf_compress, create_broadcast_secret, create_id,
|
||||
create_outgoing_rfc724_mid, create_smeared_timestamp, create_smeared_timestamps, get_abs_path,
|
||||
gm2local_offset, normalize_text, smeared_time, time, truncate_msg_text,
|
||||
create_outgoing_rfc724_mid, get_abs_path, gm2local_offset, normalize_text, time,
|
||||
truncate_msg_text,
|
||||
};
|
||||
use crate::webxdc::StatusUpdateSerial;
|
||||
|
||||
@@ -291,7 +291,7 @@ impl ChatId {
|
||||
timestamp: i64,
|
||||
) -> Result<Self> {
|
||||
let grpname = sanitize_single_line(grpname);
|
||||
let timestamp = cmp::min(timestamp, smeared_time(context));
|
||||
let timestamp = cmp::min(timestamp, time());
|
||||
let row_id =
|
||||
context.sql.insert(
|
||||
"INSERT INTO chats (type, name, name_normalized, grpid, blocked, created_timestamp, protected, param) VALUES(?, ?, ?, ?, ?, ?, 0, ?)",
|
||||
@@ -1255,7 +1255,7 @@ SELECT id, rfc724_mid, pre_rfc724_mid, timestamp, ?, 1 FROM msgs WHERE chat_id=?
|
||||
message_timestamp: i64,
|
||||
always_sort_to_bottom: bool,
|
||||
) -> Result<i64> {
|
||||
let mut sort_timestamp = cmp::min(message_timestamp, smeared_time(context));
|
||||
let mut sort_timestamp = cmp::min(message_timestamp, time());
|
||||
|
||||
let last_msg_time: Option<i64> = if always_sort_to_bottom {
|
||||
// get newest message for this chat
|
||||
@@ -2405,7 +2405,7 @@ impl ChatIdBlocked {
|
||||
_ => (),
|
||||
}
|
||||
|
||||
let smeared_time = create_smeared_timestamp(context);
|
||||
let now = time();
|
||||
|
||||
let chat_id = context
|
||||
.sql
|
||||
@@ -2420,7 +2420,7 @@ impl ChatIdBlocked {
|
||||
normalize_text(&chat_name),
|
||||
params.to_string(),
|
||||
create_blocked as u8,
|
||||
smeared_time,
|
||||
now,
|
||||
),
|
||||
)?;
|
||||
let chat_id = ChatId::new(
|
||||
@@ -2446,7 +2446,7 @@ impl ChatIdBlocked {
|
||||
&& !chat.param.exists(Param::Devicetalk)
|
||||
&& !chat.param.exists(Param::Selftalk)
|
||||
{
|
||||
chat_id.add_e2ee_notice(context, smeared_time).await?;
|
||||
chat_id.add_e2ee_notice(context, now).await?;
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
@@ -2733,7 +2733,7 @@ async fn prepare_send_msg(
|
||||
}
|
||||
msg.state = MessageState::OutPending;
|
||||
|
||||
msg.timestamp_sort = create_smeared_timestamp(context);
|
||||
msg.timestamp_sort = time();
|
||||
prepare_msg_blob(context, msg).await?;
|
||||
if !msg.hidden {
|
||||
chat_id.unarchive_if_not_muted(context, msg.state).await?;
|
||||
@@ -2907,7 +2907,7 @@ pub(crate) async fn create_send_msg_jobs(context: &Context, msg: &mut Message) -
|
||||
);
|
||||
}
|
||||
|
||||
let now = smeared_time(context);
|
||||
let now = time();
|
||||
|
||||
if rendered_msg.last_added_location_id.is_some()
|
||||
&& let Err(err) = location::set_kml_sent_timestamp(context, msg.chat_id, now).await
|
||||
@@ -2970,6 +2970,15 @@ WHERE id=?
|
||||
)?;
|
||||
for recipients_chunk in recipients.chunks(chunk_size) {
|
||||
let recipients_chunk = recipients_chunk.join(" ");
|
||||
if let Some(pre_msg) = &rendered_pre_msg {
|
||||
let row_id = stmt.execute((
|
||||
&pre_msg.rfc724_mid,
|
||||
&recipients_chunk,
|
||||
&pre_msg.message,
|
||||
msg.id,
|
||||
))?;
|
||||
row_ids.push(row_id.try_into()?);
|
||||
}
|
||||
let row_id = stmt.execute((
|
||||
&rendered_msg.rfc724_mid,
|
||||
&recipients_chunk,
|
||||
@@ -2977,16 +2986,6 @@ WHERE id=?
|
||||
msg.id,
|
||||
))?;
|
||||
row_ids.push(row_id.try_into()?);
|
||||
let Some(pre_msg) = &rendered_pre_msg else {
|
||||
continue;
|
||||
};
|
||||
let row_id = stmt.execute((
|
||||
&pre_msg.rfc724_mid,
|
||||
&recipients_chunk,
|
||||
&pre_msg.message,
|
||||
msg.id,
|
||||
))?;
|
||||
row_ids.push(row_id.try_into()?);
|
||||
}
|
||||
Ok(row_ids)
|
||||
};
|
||||
@@ -3550,7 +3549,7 @@ pub(crate) async fn create_group_ex(
|
||||
chat_name = "…".to_string();
|
||||
}
|
||||
|
||||
let timestamp = create_smeared_timestamp(context);
|
||||
let timestamp = time();
|
||||
let row_id = context
|
||||
.sql
|
||||
.insert(
|
||||
@@ -3636,7 +3635,7 @@ pub(crate) async fn create_out_broadcast_ex(
|
||||
bail!("Invalid broadcast channel name: {chat_name}.");
|
||||
}
|
||||
|
||||
let timestamp = create_smeared_timestamp(context);
|
||||
let timestamp = time();
|
||||
let trans_fn = |t: &mut rusqlite::Transaction| -> Result<ChatId> {
|
||||
let cnt: u32 = t.query_row(
|
||||
"SELECT COUNT(*) FROM chats WHERE grpid=?",
|
||||
@@ -3886,11 +3885,11 @@ pub(crate) async fn add_contact_to_chat_ex(
|
||||
return Ok(false);
|
||||
}
|
||||
if from_handshake && chat.param.get_int(Param::Unpromoted).unwrap_or_default() == 1 {
|
||||
let smeared_time = smeared_time(context);
|
||||
let now = time();
|
||||
chat.param
|
||||
.remove(Param::Unpromoted)
|
||||
.set_i64(Param::GroupNameTimestamp, smeared_time)
|
||||
.set_i64(Param::GroupDescriptionTimestamp, smeared_time);
|
||||
.set_i64(Param::GroupNameTimestamp, now)
|
||||
.set_i64(Param::GroupDescriptionTimestamp, now);
|
||||
chat.update_param(context).await?;
|
||||
}
|
||||
if context.is_self_addr(contact.get_addr()).await? {
|
||||
@@ -4453,7 +4452,6 @@ pub async fn forward_msgs(context: &Context, msg_ids: &[MsgId], chat_id: ChatId)
|
||||
}
|
||||
|
||||
/// Forwards multiple messages to a chat in another context.
|
||||
#[expect(clippy::arithmetic_side_effects)]
|
||||
pub async fn forward_msgs_2ctx(
|
||||
ctx_src: &Context,
|
||||
msg_ids: &[MsgId],
|
||||
@@ -4464,7 +4462,6 @@ pub async fn forward_msgs_2ctx(
|
||||
ensure!(!chat_id.is_special(), "can not forward to special chat");
|
||||
|
||||
let mut created_msgs: Vec<MsgId> = Vec::new();
|
||||
let mut curr_timestamp: i64;
|
||||
|
||||
chat_id
|
||||
.unarchive_if_not_muted(ctx_dst, MessageState::Undefined)
|
||||
@@ -4473,7 +4470,7 @@ pub async fn forward_msgs_2ctx(
|
||||
if let Some(reason) = chat.why_cant_send(ctx_dst).await? {
|
||||
bail!("cannot send to {chat_id}: {reason}");
|
||||
}
|
||||
curr_timestamp = create_smeared_timestamps(ctx_dst, msg_ids.len());
|
||||
let now = time();
|
||||
let mut msgs = Vec::with_capacity(msg_ids.len());
|
||||
for id in msg_ids {
|
||||
let ts: i64 = ctx_src
|
||||
@@ -4538,10 +4535,9 @@ pub async fn forward_msgs_2ctx(
|
||||
msg.state = MessageState::OutPending;
|
||||
msg.rfc724_mid = create_outgoing_rfc724_mid();
|
||||
msg.pre_rfc724_mid.clear();
|
||||
msg.timestamp_sort = curr_timestamp;
|
||||
msg.timestamp_sort = now;
|
||||
chat.prepare_msg_raw(ctx_dst, &mut msg, None).await?;
|
||||
|
||||
curr_timestamp += 1;
|
||||
if !create_send_msg_jobs(ctx_dst, &mut msg).await?.is_empty() {
|
||||
ctx_dst.scheduler.interrupt_smtp().await;
|
||||
}
|
||||
@@ -4634,7 +4630,7 @@ pub(crate) async fn save_copy_in_self_talk(
|
||||
} else {
|
||||
MessageState::InSeen
|
||||
},
|
||||
create_smeared_timestamp(context),
|
||||
time(),
|
||||
msg.param.to_string(),
|
||||
src_msg_id,
|
||||
src_msg_id,
|
||||
@@ -4811,7 +4807,7 @@ pub async fn add_device_msg_with_importance(
|
||||
chat_id = ChatId::get_for_contact(context, ContactId::DEVICE).await?;
|
||||
|
||||
let rfc724_mid = create_outgoing_rfc724_mid();
|
||||
let timestamp_sent = create_smeared_timestamp(context);
|
||||
let timestamp_sent = time();
|
||||
|
||||
// makes sure, the added message is the last one,
|
||||
// even if the date is wrong (useful esp. when warning about bad dates)
|
||||
@@ -4958,7 +4954,7 @@ pub(crate) async fn add_info_msg_with_cmd(
|
||||
} else {
|
||||
let sort_to_bottom = true;
|
||||
chat_id
|
||||
.calc_sort_timestamp(context, smeared_time(context), sort_to_bottom)
|
||||
.calc_sort_timestamp(context, time(), sort_to_bottom)
|
||||
.await?
|
||||
};
|
||||
|
||||
@@ -5121,7 +5117,7 @@ async fn set_contacts_by_fingerprints(
|
||||
Ok(broadcast_contacts_added)
|
||||
})
|
||||
.await?;
|
||||
let timestamp = smeared_time(context);
|
||||
let timestamp = time();
|
||||
for added_id in broadcast_contacts_added {
|
||||
let msg = stock_str::msg_add_member_local(context, added_id, ContactId::UNDEFINED).await;
|
||||
add_info_msg_with_cmd(
|
||||
|
||||
@@ -1625,6 +1625,7 @@ async fn test_set_chat_name() {
|
||||
"another name",
|
||||
"something different",
|
||||
] {
|
||||
SystemTime::shift(Duration::from_secs(1));
|
||||
set_chat_name(alice, chat_id, new_name).await.unwrap();
|
||||
let sent_msg = alice.pop_sent_msg().await;
|
||||
let received_msg = bob.recv_msg(&sent_msg).await;
|
||||
@@ -3437,8 +3438,9 @@ async fn test_chat_description(
|
||||
"",
|
||||
"ä ẟ 😂",
|
||||
] {
|
||||
SystemTime::shift(Duration::from_secs(1));
|
||||
tcm.section(&format!(
|
||||
"Alice sets the chat description to '{description}'"
|
||||
"Alice sets the chat description to {description:?}"
|
||||
));
|
||||
set_chat_description(alice, alice_chat_id, description).await?;
|
||||
let sent = alice.pop_sent_msg().await;
|
||||
@@ -4459,7 +4461,9 @@ async fn test_get_chat_media_webxdc_order() -> Result<()> {
|
||||
assert_eq!(media.first().unwrap(), &instance1_id);
|
||||
assert_eq!(media.get(1).unwrap(), &instance2_id);
|
||||
|
||||
// add a status update for the oder instance; that resorts the list
|
||||
SystemTime::shift(Duration::from_secs(1));
|
||||
|
||||
// add a status update for the other instance; that resorts the list
|
||||
alice
|
||||
.send_webxdc_status_update(instance1_id, r#"{"payload": {"foo": "bar"}}"#)
|
||||
.await?;
|
||||
@@ -4873,10 +4877,6 @@ async fn test_sync_broadcast_and_send_message() -> Result<()> {
|
||||
vec![a2b_contact_id]
|
||||
);
|
||||
|
||||
// alice2's smeared clock may be behind alice1's one, so we need to work around "hi" appearing
|
||||
// before "You joined the channel." for bob. alice1 makes 3 more calls of
|
||||
// create_smeared_timestamp() than alice2 does as of 2026-03-10.
|
||||
SystemTime::shift(Duration::from_secs(3));
|
||||
tcm.section("Alice's second device sends a message to the channel");
|
||||
let sent_msg = alice2.send_text(a2_broadcast_id, "hi").await;
|
||||
let msg = bob.recv_msg(&sent_msg).await;
|
||||
|
||||
@@ -32,7 +32,6 @@ use crate::quota::QuotaInfo;
|
||||
use crate::scheduler::{ConnectivityStore, SchedulerState};
|
||||
use crate::sql::Sql;
|
||||
use crate::stock_str::StockStrings;
|
||||
use crate::timesmearing::SmearedTimestamp;
|
||||
use crate::tools::{self, duration_to_str, time, time_elapsed};
|
||||
use crate::transport::ConfiguredLoginParam;
|
||||
use crate::{chatlist_events, stats};
|
||||
@@ -228,7 +227,6 @@ pub struct InnerContext {
|
||||
/// Blob directory path
|
||||
pub(crate) blobdir: PathBuf,
|
||||
pub(crate) sql: Sql,
|
||||
pub(crate) smeared_timestamp: SmearedTimestamp,
|
||||
/// The global "ongoing" process state.
|
||||
///
|
||||
/// This is a global mutex-like state for operations which should be modal in the
|
||||
@@ -498,7 +496,6 @@ impl Context {
|
||||
blobdir,
|
||||
running_state: RwLock::new(Default::default()),
|
||||
sql: Sql::new(dbfile),
|
||||
smeared_timestamp: SmearedTimestamp::new(),
|
||||
oauth2_mutex: Mutex::new(()),
|
||||
wrong_pw_warning_mutex: Mutex::new(()),
|
||||
housekeeping_mutex: Mutex::new(()),
|
||||
|
||||
@@ -10,7 +10,6 @@ use crate::context::Context;
|
||||
use crate::imap::session::Session;
|
||||
use crate::log::warn;
|
||||
use crate::message::{self, Message, MsgId, rfc724_mid_exists};
|
||||
use crate::tools;
|
||||
use crate::{EventType, chatlist_events};
|
||||
|
||||
pub(crate) mod post_msg_metadata;
|
||||
@@ -321,22 +320,12 @@ pub(crate) async fn download_known_post_messages_without_pre_message(
|
||||
context: &Context,
|
||||
session: &mut Session,
|
||||
) -> Result<()> {
|
||||
const PRE_MSG_WAIT_TIME: i64 = 30;
|
||||
let now = tools::time();
|
||||
let rfc724_mids = context
|
||||
.sql
|
||||
.query_map_vec(
|
||||
"
|
||||
SELECT rfc724_mid FROM available_post_msgs
|
||||
WHERE timestamp<=? OR timestamp>?
|
||||
ORDER BY timestamp, rowid
|
||||
",
|
||||
(now.saturating_sub(PRE_MSG_WAIT_TIME), now),
|
||||
|row| {
|
||||
let rfc724_mid: String = row.get(0)?;
|
||||
Ok(rfc724_mid)
|
||||
},
|
||||
)
|
||||
.query_map_vec("SELECT rfc724_mid FROM available_post_msgs", (), |row| {
|
||||
let rfc724_mid: String = row.get(0)?;
|
||||
Ok(rfc724_mid)
|
||||
})
|
||||
.await?;
|
||||
for rfc724_mid in &rfc724_mids {
|
||||
if msg_is_downloaded_for(context, rfc724_mid).await? {
|
||||
|
||||
@@ -10,7 +10,6 @@ use crate::location;
|
||||
use crate::message::markseen_msgs;
|
||||
use crate::receive_imf::receive_imf;
|
||||
use crate::test_utils::{TestContext, TestContextManager};
|
||||
use crate::timesmearing::MAX_SECONDS_TO_LEND_FROM_FUTURE;
|
||||
use crate::{
|
||||
chat::{self, Chat, ChatItem, create_group, send_text_msg},
|
||||
tools::IsNoneOrEmpty,
|
||||
@@ -352,17 +351,9 @@ async fn test_ephemeral_delete_msgs() -> Result<()> {
|
||||
let now = time();
|
||||
let msg = t.send_text(bob_chat.id, "Message text").await;
|
||||
|
||||
check_msg_will_be_deleted(
|
||||
&t,
|
||||
msg.sender_msg_id,
|
||||
&bob_chat,
|
||||
now + 1799,
|
||||
// The message may appear to be sent MAX_SECONDS_TO_LEND_FROM_FUTURE later and
|
||||
// therefore be deleted MAX_SECONDS_TO_LEND_FROM_FUTURE later.
|
||||
time() + 1801 + MAX_SECONDS_TO_LEND_FROM_FUTURE,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
check_msg_will_be_deleted(&t, msg.sender_msg_id, &bob_chat, now + 1799, time() + 1801)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Enable ephemeral messages with Bob -> message will be deleted after 60s.
|
||||
// This tests that the message is deleted at min(ephemeral deletion time, DeleteDeviceAfter deletion time).
|
||||
|
||||
11
src/imap.rs
11
src/imap.rs
@@ -601,7 +601,7 @@ impl Imap {
|
||||
let _fetch_msgs_lock_guard = context.fetch_msgs_mutex.lock().await;
|
||||
|
||||
let mut uids_fetch: Vec<u32> = Vec::new();
|
||||
let mut available_post_msgs: Vec<(String, i64)> = Vec::new();
|
||||
let mut available_post_msgs: Vec<String> = Vec::new();
|
||||
let mut download_later: Vec<String> = Vec::new();
|
||||
let mut uid_message_ids = BTreeMap::new();
|
||||
let mut largest_uid_skipped = None;
|
||||
@@ -689,7 +689,7 @@ impl Imap {
|
||||
.is_some()
|
||||
{
|
||||
info!(context, "{message_id:?} is a post-message.");
|
||||
available_post_msgs.push((message_id.clone(), time()));
|
||||
available_post_msgs.push(message_id.clone());
|
||||
|
||||
let is_bot = context.get_config_bool(Config::Bot).await?;
|
||||
if is_bot && download_limit.is_none_or(|download_limit| size <= download_limit)
|
||||
@@ -793,10 +793,9 @@ impl Imap {
|
||||
download_later.len(),
|
||||
);
|
||||
let trans_fn = |t: &mut rusqlite::Transaction| {
|
||||
let mut stmt =
|
||||
t.prepare("INSERT OR IGNORE INTO available_post_msgs VALUES (?,?)")?;
|
||||
for entry in available_post_msgs {
|
||||
stmt.execute(entry)
|
||||
let mut stmt = t.prepare("INSERT OR IGNORE INTO available_post_msgs VALUES (?)")?;
|
||||
for rfc724_mid in available_post_msgs {
|
||||
stmt.execute((rfc724_mid,))
|
||||
.context("INSERT OR IGNORE INTO available_post_msgs")?;
|
||||
}
|
||||
let mut stmt =
|
||||
|
||||
@@ -94,7 +94,6 @@ mod smtp;
|
||||
pub mod stock_str;
|
||||
pub mod storage_usage;
|
||||
mod sync;
|
||||
mod timesmearing;
|
||||
mod token;
|
||||
mod transport;
|
||||
mod update_helper;
|
||||
|
||||
@@ -35,10 +35,7 @@ use crate::peer_channels::{create_iroh_header, get_iroh_topic_for_msg};
|
||||
use crate::pgp::{SeipdVersion, addresses_from_public_key, pubkey_supports_seipdv2};
|
||||
use crate::simplify::escape_message_footer_marks;
|
||||
use crate::stock_str;
|
||||
use crate::tools::{
|
||||
IsNoneOrEmpty, create_outgoing_rfc724_mid, create_smeared_timestamp, remove_subject_prefix,
|
||||
time,
|
||||
};
|
||||
use crate::tools::{IsNoneOrEmpty, create_outgoing_rfc724_mid, remove_subject_prefix, time};
|
||||
use crate::webxdc::StatusUpdateSerial;
|
||||
|
||||
// attachments of 25 mb brutto should work on the majority of providers
|
||||
@@ -580,7 +577,7 @@ impl MimeFactory {
|
||||
) -> Result<MimeFactory> {
|
||||
let contact = Contact::get_by_id(context, from_id).await?;
|
||||
let from_addr = context.get_primary_self_addr().await?;
|
||||
let timestamp = create_smeared_timestamp(context);
|
||||
let timestamp = time();
|
||||
|
||||
let addr = contact.get_addr().to_string();
|
||||
let encryption_pubkeys = if from_id == ContactId::SELF {
|
||||
@@ -2301,7 +2298,7 @@ pub(crate) async fn render_symm_encrypted_securejoin_message(
|
||||
mail_builder::headers::text::Text::new("Secure-Join".to_string()).into(),
|
||||
));
|
||||
|
||||
let timestamp = create_smeared_timestamp(context);
|
||||
let timestamp = time();
|
||||
let date = chrono::DateTime::<chrono::Utc>::from_timestamp(timestamp, 0)
|
||||
.unwrap()
|
||||
.to_rfc2822();
|
||||
|
||||
@@ -31,9 +31,7 @@ use crate::message::{self, Message, MsgId, Viewtype, get_vcard_summary, set_msg_
|
||||
use crate::param::{Param, Params};
|
||||
use crate::simplify::{SimplifiedText, simplify};
|
||||
use crate::sync::SyncItems;
|
||||
use crate::tools::{
|
||||
get_filemeta, parse_receive_headers, smeared_time, time, truncate_msg_text, validate_id,
|
||||
};
|
||||
use crate::tools::{get_filemeta, parse_receive_headers, time, truncate_msg_text, validate_id};
|
||||
use crate::{chatlist_events, location, tools};
|
||||
|
||||
/// Public key extracted from `Autocrypt-Gossip`
|
||||
@@ -271,7 +269,7 @@ impl MimeMessage {
|
||||
pub(crate) async fn from_bytes(context: &Context, body: &[u8]) -> Result<Self> {
|
||||
let mail = mailparse::parse_mail(body)?;
|
||||
|
||||
let timestamp_rcvd = smeared_time(context);
|
||||
let timestamp_rcvd = time();
|
||||
let mut timestamp_sent =
|
||||
Self::get_timestamp_sent(&mail.headers, timestamp_rcvd, timestamp_rcvd);
|
||||
let hop_info = parse_receive_headers(&mail.get_headers());
|
||||
|
||||
@@ -525,10 +525,18 @@ pub(crate) async fn receive_imf_inner(
|
||||
"Receiving message {rfc724_mid_orig:?}, seen={seen}...",
|
||||
);
|
||||
|
||||
let msg_id = message::rfc724_mid_exists(context, rfc724_mid_orig).await?;
|
||||
if let Some(msg_id) = msg_id
|
||||
&& !mime_parser.incoming
|
||||
{
|
||||
// These checks must be done before processing of SecureJoin and other special messages.
|
||||
if mime_parser.pre_message == mimeparser::PreMessageMode::Post {
|
||||
// Post-Message just replaces the attachment and modifies Params, not the whole message.
|
||||
// This is done in the `handle_post_message` method.
|
||||
} else if let Some(msg_id) = message::rfc724_mid_exists(context, rfc724_mid_orig).await? {
|
||||
info!(
|
||||
context,
|
||||
"Message {rfc724_mid} is already in some chat or deleted."
|
||||
);
|
||||
if mime_parser.incoming {
|
||||
return Ok(None);
|
||||
}
|
||||
// For the case if we missed a successful SMTP response. Be optimistic that the message is
|
||||
// delivered also.
|
||||
let self_addr = context.get_primary_self_addr().await?;
|
||||
@@ -543,16 +551,6 @@ pub(crate) async fn receive_imf_inner(
|
||||
if !msg_has_pending_smtp_job(context, msg_id).await? {
|
||||
msg_id.set_delivered(context).await?;
|
||||
}
|
||||
}
|
||||
// These checks must be done before processing of SecureJoin and other special messages.
|
||||
if mime_parser.pre_message == mimeparser::PreMessageMode::Post {
|
||||
// Post-Message just replaces the attachment and modifies Params, not the whole message.
|
||||
// This is done in the `handle_post_message` method.
|
||||
} else if msg_id.is_some() {
|
||||
info!(
|
||||
context,
|
||||
"Message {rfc724_mid} is already in some chat or deleted."
|
||||
);
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
@@ -2506,10 +2504,7 @@ WHERE id=?
|
||||
part.typ,
|
||||
part.bytes as isize,
|
||||
part.error.as_deref().unwrap_or_default(),
|
||||
match mime_parser.incoming {
|
||||
true => state,
|
||||
false => MessageState::Undefined,
|
||||
},
|
||||
state,
|
||||
DownloadState::Done as u32,
|
||||
original_msg.id,
|
||||
),
|
||||
|
||||
@@ -3975,6 +3975,8 @@ async fn test_delayed_removal_is_ignored() -> Result<()> {
|
||||
remove_contact_from_chat(bob, bob_chat_id, bob_contact_fiona).await?;
|
||||
let remove_msg = bob.pop_sent_msg().await;
|
||||
|
||||
SystemTime::shift(Duration::from_secs(1));
|
||||
|
||||
// Bob adds new members Dom and Elena, but first addition message is lost.
|
||||
let dom = &tcm.dom().await;
|
||||
let elena = &tcm.elena().await;
|
||||
@@ -3991,6 +3993,8 @@ async fn test_delayed_removal_is_ignored() -> Result<()> {
|
||||
alice.recv_msg(&add_msg).await;
|
||||
assert_eq!(get_chat_contacts(alice, chat_id).await?.len(), 4);
|
||||
|
||||
SystemTime::shift(Duration::from_secs(1));
|
||||
|
||||
// Alice re-adds Fiona.
|
||||
add_contact_to_chat(alice, chat_id, alice_fiona).await?;
|
||||
assert_eq!(get_chat_contacts(alice, chat_id).await?.len(), 5);
|
||||
@@ -5592,27 +5596,27 @@ async fn test_mark_message_as_delivered_only_after_sent_out_fully() -> Result<()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let (post_msg_id, post_msg_payload) = first_row_in_smtp_queue(alice).await;
|
||||
assert_eq!(msg_id, post_msg_id);
|
||||
assert!(post_msg_payload.len() > file_bytes.len());
|
||||
|
||||
assert_eq!(msg_id.get_state(alice).await?, MessageState::OutPending);
|
||||
// Alice receives her own post-message because of bcc_self
|
||||
// This should not yet mark the message as delivered,
|
||||
// because not everything was sent,
|
||||
// but it does remove the post-message from the SMTP queue.
|
||||
receive_imf(alice, post_msg_payload.as_bytes(), false).await?;
|
||||
assert_eq!(msg_id.get_state(alice).await?, MessageState::OutPending);
|
||||
|
||||
let (pre_msg_id, pre_msg_payload) = first_row_in_smtp_queue(alice).await;
|
||||
assert_eq!(msg_id, pre_msg_id);
|
||||
assert!(pre_msg_payload.len() < file_bytes.len());
|
||||
|
||||
assert_eq!(msg_id.get_state(alice).await?, MessageState::OutPending);
|
||||
// Alice receives her own pre-message because of bcc_self
|
||||
// This should not yet mark the message as delivered,
|
||||
// because not everything was sent,
|
||||
// but it does remove the pre-message from the SMTP queue
|
||||
receive_imf(alice, pre_msg_payload.as_bytes(), false).await?;
|
||||
assert_eq!(msg_id.get_state(alice).await?, MessageState::OutPending);
|
||||
|
||||
let (post_msg_id, post_msg_payload) = first_row_in_smtp_queue(alice).await;
|
||||
assert_eq!(msg_id, post_msg_id);
|
||||
assert!(post_msg_payload.len() > file_bytes.len());
|
||||
|
||||
assert_eq!(msg_id.get_state(alice).await?, MessageState::OutPending);
|
||||
// Alice receives her own post-message because of bcc_self
|
||||
// This should now mark the message as delivered,
|
||||
// because everything was sent by now.
|
||||
receive_imf(alice, pre_msg_payload.as_bytes(), false).await?;
|
||||
receive_imf(alice, post_msg_payload.as_bytes(), false).await?;
|
||||
assert_eq!(msg_id.get_state(alice).await?, MessageState::OutDelivered);
|
||||
|
||||
Ok(())
|
||||
|
||||
@@ -19,7 +19,7 @@ use crate::securejoin::{
|
||||
};
|
||||
use crate::stock_str;
|
||||
use crate::sync::Sync::*;
|
||||
use crate::tools::{create_outgoing_rfc724_mid, smeared_time, time};
|
||||
use crate::tools::{create_outgoing_rfc724_mid, time};
|
||||
use crate::{chatlist_events, mimefactory};
|
||||
|
||||
/// Starts the securejoin protocol with the QR `invite`.
|
||||
@@ -465,7 +465,7 @@ async fn joining_chat_id(
|
||||
name,
|
||||
Blocked::Not,
|
||||
None,
|
||||
smeared_time(context),
|
||||
time(),
|
||||
)
|
||||
.await?
|
||||
}
|
||||
|
||||
@@ -379,7 +379,7 @@ pub(crate) async fn send_msg_to_smtp(
|
||||
if retries > 6 {
|
||||
context
|
||||
.sql
|
||||
.execute("DELETE FROM smtp WHERE msg_id=?", (msg_id,))
|
||||
.execute("DELETE FROM smtp WHERE id=?", (rowid,))
|
||||
.await
|
||||
.context("Failed to remove message with exceeded retry limit from smtp table")?;
|
||||
if let Some(mut msg) = Message::load_from_db_optional(context, msg_id).await? {
|
||||
|
||||
@@ -2385,19 +2385,6 @@ UPDATE msgs SET state=19 WHERE state=24; -- Change OutPreparing to OutFailed.
|
||||
.await?;
|
||||
}
|
||||
|
||||
inc_and_check(&mut migration_version, 153)?;
|
||||
if dbversion < migration_version {
|
||||
sql.execute_migration(
|
||||
"
|
||||
CREATE INDEX smtp_index_msg_id ON smtp (msg_id, id);
|
||||
ALTER TABLE available_post_msgs ADD COLUMN timestamp INTEGER DEFAULT 0 NOT NULL;
|
||||
CREATE INDEX available_post_msgs_timestamp ON available_post_msgs (timestamp);
|
||||
",
|
||||
migration_version,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
let new_version = sql
|
||||
.get_raw_config_int(VERSION_CFG)
|
||||
.await?
|
||||
|
||||
@@ -722,14 +722,12 @@ ORDER BY id"
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns `SentMessage` instances representing `smtp` rows for the given message. Returned
|
||||
/// items go in reverse order for historical reasons.
|
||||
pub async fn get_smtp_rows_for_msg<'a>(&'a self, msg_id: MsgId) -> Vec<SentMessage<'a>> {
|
||||
let sent_msgs = self
|
||||
.ctx
|
||||
.sql
|
||||
.query_map_vec(
|
||||
"SELECT id, msg_id, mime, recipients FROM smtp WHERE msg_id=? ORDER BY id DESC",
|
||||
"SELECT id, msg_id, mime, recipients FROM smtp WHERE msg_id=?",
|
||||
(msg_id,),
|
||||
|row| {
|
||||
let _id: MsgId = row.get(0)?;
|
||||
@@ -1057,17 +1055,9 @@ ORDER BY id"
|
||||
/// This is not hooked up to any SMTP-IMAP pipeline, so the other account must call
|
||||
/// [`TestContext::recv_msg`] with the returned [`SentMessage`] if it wants to receive
|
||||
/// the message.
|
||||
///
|
||||
/// Removes SMTP jobs existed before and marks the corresponding messages as delivered, as
|
||||
/// tracking of these jobs is probably already lost by the test code.
|
||||
pub async fn send_msg(&self, chat_id: ChatId, msg: &mut Message) -> SentMessage<'_> {
|
||||
while self.pop_sent_msg_opt(Duration::ZERO).await.is_some() {}
|
||||
let msg_id = chat::send_msg(self, chat_id, msg).await.unwrap();
|
||||
let rev_order = false;
|
||||
let res = self
|
||||
.pop_sent_msg_ex(rev_order, Duration::ZERO)
|
||||
.await
|
||||
.unwrap();
|
||||
let res = self.pop_sent_msg().await;
|
||||
assert_eq!(
|
||||
res.sender_msg_id, msg_id,
|
||||
"Apparently the message was not actually sent out"
|
||||
|
||||
@@ -106,7 +106,7 @@ async fn test_receive_both() -> Result<()> {
|
||||
assert_eq!(msg.text, "test".to_owned());
|
||||
|
||||
forward_msgs(alice, &[alice_msg_id], alice_chat_id).await?;
|
||||
let rev_order = true;
|
||||
let rev_order = false;
|
||||
let msg = bob
|
||||
.recv_msg(
|
||||
&alice
|
||||
|
||||
@@ -547,8 +547,8 @@ async fn test_webxdc_updates_in_post_message_after_pre_message() -> Result<()> {
|
||||
.await?;
|
||||
|
||||
send_msg(alice, alice_chat_id, &mut alice_instance).await?;
|
||||
let pre_message = alice.pop_sent_msg().await;
|
||||
let post_message = alice.pop_sent_msg().await;
|
||||
let pre_message = alice.pop_sent_msg().await;
|
||||
|
||||
let bob_instance = bob.recv_msg(&pre_message).await;
|
||||
assert_eq!(bob_instance.download_state, DownloadState::Available);
|
||||
@@ -588,8 +588,8 @@ async fn test_webxdc_updates_in_post_message_without_pre_message() -> Result<()>
|
||||
.await?;
|
||||
|
||||
send_msg(alice, alice_chat_id, &mut alice_instance).await?;
|
||||
let pre_message = alice.pop_sent_msg().await;
|
||||
let post_message = alice.pop_sent_msg().await;
|
||||
let pre_message = alice.pop_sent_msg().await;
|
||||
|
||||
// Bob receives post-message first.
|
||||
let bob_instance = bob.recv_msg(&post_message).await;
|
||||
|
||||
@@ -1,194 +0,0 @@
|
||||
//! # Time smearing.
|
||||
//!
|
||||
//! As e-mails typically only use a second-based-resolution for timestamps,
|
||||
//! the order of two mails sent within one second is unclear.
|
||||
//! This is bad e.g. when forwarding some messages from a chat -
|
||||
//! these messages will appear at the recipient easily out of order.
|
||||
//!
|
||||
//! We work around this issue by not sending out two mails with the same timestamp.
|
||||
//! For this purpose, in short, we track the last timestamp used in `last_smeared_timestamp`
|
||||
//! when another timestamp is needed in the same second, we use `last_smeared_timestamp+1`
|
||||
//! after some moments without messages sent out,
|
||||
//! `last_smeared_timestamp` is again in sync with the normal time.
|
||||
//!
|
||||
//! However, we do not do all this for the far future,
|
||||
//! but at max `MAX_SECONDS_TO_LEND_FROM_FUTURE`
|
||||
|
||||
use std::cmp::{max, min};
|
||||
use std::sync::atomic::{AtomicI64, Ordering};
|
||||
|
||||
pub(crate) const MAX_SECONDS_TO_LEND_FROM_FUTURE: i64 = 30;
|
||||
|
||||
/// Smeared timestamp generator.
|
||||
#[derive(Debug)]
|
||||
pub struct SmearedTimestamp {
|
||||
/// Next timestamp available for allocation.
|
||||
smeared_timestamp: AtomicI64,
|
||||
}
|
||||
|
||||
impl SmearedTimestamp {
|
||||
/// Creates a new smeared timestamp generator.
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
smeared_timestamp: AtomicI64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
/// Allocates `count` unique timestamps.
|
||||
///
|
||||
/// Returns the first allocated timestamp.
|
||||
#[expect(clippy::arithmetic_side_effects)]
|
||||
pub fn create_n(&self, now: i64, count: i64) -> i64 {
|
||||
let mut prev = self.smeared_timestamp.load(Ordering::Relaxed);
|
||||
loop {
|
||||
// Advance the timestamp if it is in the past,
|
||||
// but keep `count - 1` timestamps from the past if possible.
|
||||
let t = max(prev, now - count + 1);
|
||||
|
||||
// Rewind the time back if there is no room
|
||||
// to allocate `count` timestamps without going too far into the future.
|
||||
// Not going too far into the future
|
||||
// is more important than generating unique timestamps.
|
||||
let first = min(t, now + MAX_SECONDS_TO_LEND_FROM_FUTURE - count + 1);
|
||||
|
||||
// Allocate `count` timestamps by advancing the current timestamp.
|
||||
let next = first + count;
|
||||
|
||||
if let Err(x) = self.smeared_timestamp.compare_exchange_weak(
|
||||
prev,
|
||||
next,
|
||||
Ordering::Relaxed,
|
||||
Ordering::Relaxed,
|
||||
) {
|
||||
prev = x;
|
||||
} else {
|
||||
return first;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Creates a single timestamp.
|
||||
pub fn create(&self, now: i64) -> i64 {
|
||||
self.create_n(now, 1)
|
||||
}
|
||||
|
||||
/// Returns the current smeared timestamp.
|
||||
pub fn current(&self) -> i64 {
|
||||
self.smeared_timestamp.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::test_utils::TestContext;
|
||||
use crate::tools::{
|
||||
SystemTime, create_smeared_timestamp, create_smeared_timestamps, smeared_time, time,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn test_smeared_timestamp() {
|
||||
let smeared_timestamp = SmearedTimestamp::new();
|
||||
let now = time();
|
||||
|
||||
assert_eq!(smeared_timestamp.current(), 0);
|
||||
|
||||
for i in 0..MAX_SECONDS_TO_LEND_FROM_FUTURE {
|
||||
assert_eq!(smeared_timestamp.create(now), now + i);
|
||||
}
|
||||
assert_eq!(
|
||||
smeared_timestamp.create(now),
|
||||
now + MAX_SECONDS_TO_LEND_FROM_FUTURE
|
||||
);
|
||||
assert_eq!(
|
||||
smeared_timestamp.create(now),
|
||||
now + MAX_SECONDS_TO_LEND_FROM_FUTURE
|
||||
);
|
||||
|
||||
// System time rewinds back by 1000 seconds.
|
||||
let now = now - 1000;
|
||||
assert_eq!(
|
||||
smeared_timestamp.create(now),
|
||||
now + MAX_SECONDS_TO_LEND_FROM_FUTURE
|
||||
);
|
||||
assert_eq!(
|
||||
smeared_timestamp.create(now),
|
||||
now + MAX_SECONDS_TO_LEND_FROM_FUTURE
|
||||
);
|
||||
assert_eq!(
|
||||
smeared_timestamp.create(now + 1),
|
||||
now + MAX_SECONDS_TO_LEND_FROM_FUTURE + 1
|
||||
);
|
||||
assert_eq!(smeared_timestamp.create(now + 100), now + 100);
|
||||
assert_eq!(smeared_timestamp.create(now + 100), now + 101);
|
||||
assert_eq!(smeared_timestamp.create(now + 100), now + 102);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_create_n_smeared_timestamps() {
|
||||
let smeared_timestamp = SmearedTimestamp::new();
|
||||
let now = time();
|
||||
|
||||
// Create a single timestamp to initialize the generator.
|
||||
assert_eq!(smeared_timestamp.create(now), now);
|
||||
|
||||
// Wait a minute.
|
||||
let now = now + 60;
|
||||
|
||||
// Simulate forwarding 7 messages.
|
||||
let forwarded_messages = 7;
|
||||
|
||||
// We have not sent anything for a minute,
|
||||
// so we can take the current timestamp and take 6 timestamps from the past.
|
||||
assert_eq!(smeared_timestamp.create_n(now, forwarded_messages), now - 6);
|
||||
|
||||
assert_eq!(smeared_timestamp.current(), now + 1);
|
||||
|
||||
// Wait 4 seconds.
|
||||
// Now we have 3 free timestamps in the past.
|
||||
let now = now + 4;
|
||||
|
||||
assert_eq!(smeared_timestamp.current(), now - 3);
|
||||
|
||||
// Forward another 7 messages.
|
||||
// We can only lend 3 timestamps from the past.
|
||||
assert_eq!(smeared_timestamp.create_n(now, forwarded_messages), now - 3);
|
||||
|
||||
// We had to borrow 3 timestamps from the future
|
||||
// because there were not enough timestamps in the past.
|
||||
assert_eq!(smeared_timestamp.current(), now + 4);
|
||||
|
||||
// Forward another 32 messages.
|
||||
// We cannot use more than 30 timestamps from the future,
|
||||
// so we use 30 timestamps from the future,
|
||||
// the current timestamp and one timestamp from the past.
|
||||
assert_eq!(smeared_timestamp.create_n(now, 32), now - 1);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_create_smeared_timestamp() {
|
||||
let t = TestContext::new().await;
|
||||
assert_ne!(create_smeared_timestamp(&t), create_smeared_timestamp(&t));
|
||||
assert!(
|
||||
create_smeared_timestamp(&t)
|
||||
>= SystemTime::now()
|
||||
.duration_since(SystemTime::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs() as i64
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn test_create_smeared_timestamps() {
|
||||
let t = TestContext::new().await;
|
||||
let count = MAX_SECONDS_TO_LEND_FROM_FUTURE - 1;
|
||||
let start = create_smeared_timestamps(&t, count as usize);
|
||||
let next = smeared_time(&t);
|
||||
assert!((start + count - 1) < next);
|
||||
|
||||
let count = MAX_SECONDS_TO_LEND_FROM_FUTURE + 30;
|
||||
let start = create_smeared_timestamps(&t, count as usize);
|
||||
let next = smeared_time(&t);
|
||||
assert!((start + count - 1) < next);
|
||||
}
|
||||
}
|
||||
23
src/tools.rs
23
src/tools.rs
@@ -180,29 +180,6 @@ pub(crate) fn gm2local_offset() -> i64 {
|
||||
i64::from(lt.offset().local_minus_utc())
|
||||
}
|
||||
|
||||
/// Returns the current smeared timestamp,
|
||||
///
|
||||
/// The returned timestamp MAY NOT be unique and MUST NOT go to "Date" header.
|
||||
pub(crate) fn smeared_time(context: &Context) -> i64 {
|
||||
let now = time();
|
||||
let ts = context.smeared_timestamp.current();
|
||||
std::cmp::max(ts, now)
|
||||
}
|
||||
|
||||
/// Returns a timestamp that is guaranteed to be unique.
|
||||
pub(crate) fn create_smeared_timestamp(context: &Context) -> i64 {
|
||||
let now = time();
|
||||
context.smeared_timestamp.create(now)
|
||||
}
|
||||
|
||||
// creates `count` timestamps that are guaranteed to be unique.
|
||||
// the first created timestamps is returned directly,
|
||||
// get the other timestamps just by adding 1..count-1
|
||||
pub(crate) fn create_smeared_timestamps(context: &Context, count: usize) -> i64 {
|
||||
let now = time();
|
||||
context.smeared_timestamp.create_n(now, count as i64)
|
||||
}
|
||||
|
||||
/// Returns the last release timestamp as a unix timestamp compatible for comparison with time() and
|
||||
/// database times.
|
||||
pub fn get_release_timestamp() -> i64 {
|
||||
|
||||
@@ -46,7 +46,7 @@ use crate::mimefactory::RECOMMENDED_FILE_SIZE;
|
||||
use crate::mimeparser::SystemMessage;
|
||||
use crate::param::Param;
|
||||
use crate::param::Params;
|
||||
use crate::tools::{create_id, create_smeared_timestamp, get_abs_path};
|
||||
use crate::tools::{create_id, get_abs_path, time};
|
||||
|
||||
/// The current API version.
|
||||
/// If `min_api` in manifest.toml is set to a larger value,
|
||||
@@ -558,7 +558,7 @@ impl Context {
|
||||
.create_status_update_record(
|
||||
&instance,
|
||||
status_update,
|
||||
create_smeared_timestamp(self),
|
||||
time(),
|
||||
send_now,
|
||||
ContactId::SELF,
|
||||
)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
OutBroadcast#Chat#1001: Channel [0 member(s)]
|
||||
--------------------------------------------------------------------------------
|
||||
Msg#1001: info (Contact#Contact#Info): Messages are end-to-end encrypted. [NOTICED][INFO]
|
||||
Msg#1008🔒: Me (Contact#Contact#Self): hi √
|
||||
Msg#1006🔒: Me (Contact#Contact#Self): Member bob@example.net added. [INFO] √
|
||||
Msg#1008🔒: Me (Contact#Contact#Self): hi √
|
||||
Msg#1009🔒: Me (Contact#Contact#Self): You removed member bob@example.net. [INFO] √
|
||||
--------------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user