filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Distributed SMTP queues (untested)

Browse Source
This commit is contained in:
Mauro D
2024-02-08 20:03:57 -03:00
parent d15f598460
commit d16119f54b
60 changed files with 2990 additions and 3828 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
crates/smtp/src/scripts/event_loop.rs
View File

@@ -21,7 +21,7 @@
* for more details.
*/
use std::{sync::Arc, time::Duration};
use std::sync::Arc;
use mail_auth::common::headers::HeaderWriter;
use sieve::{
@@ -32,18 +32,12 @@ use smtp_proto::{
MAIL_BY_TRACE, MAIL_RET_FULL, MAIL_RET_HDRS, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE,
RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS,
};
use store::{backend::memory::MemoryStore, LookupKey, LookupStore, LookupValue};
use store::{backend::memory::MemoryStore, LookupStore};
use tokio::runtime::Handle;
use crate::{
core::SMTP,
queue::{DomainPart, InstantFromTimestamp, Message},
};
use crate::{core::SMTP, queue::DomainPart};
use super::{
plugins::{lookup::VariableExists, PluginContext},
ScriptModification, ScriptParameters, ScriptResult,
};
use super::{plugins::PluginContext, ScriptModification, ScriptParameters, ScriptResult};
impl SMTP {
pub fn run_script_blocking(
@@ -97,15 +91,15 @@ impl SMTP {
'outer: for list in lists {
if let Some(store) = self.shared.lookup_stores.get(&list) {
for value in &values {
if let Ok(LookupValue::Value { .. }) = handle.block_on(
store.key_get::<VariableExists>(LookupKey::Key(
if let Ok(true) = handle.block_on(
store.key_exists(
if !matches!(match_as, MatchAs::Lowercase) {
value.clone()
} else {
value.to_lowercase()
}
.into_bytes(),
)),
),
) {
input = true.into();
break 'outer;
@@ -156,7 +150,7 @@ impl SMTP {
// Build message
let return_path_lcase = self.sieve.return_path.to_lowercase();
let return_path_domain = return_path_lcase.domain_part().to_string();
let mut message = Message::new_boxed(
let mut message = self.queue.new_message(
self.sieve.return_path.clone(),
return_path_lcase,
return_path_domain,
@@ -223,7 +217,6 @@ impl SMTP {
if trace {
message.flags |= MAIL_BY_TRACE;
}
let rlimit = Duration::from_secs(rlimit);
match mode {
ByMode::Notify => {
for domain in &mut message.domains {
@@ -246,16 +239,15 @@ impl SMTP {
if trace {
message.flags |= MAIL_BY_TRACE;
}
let alimit = (alimit as u64).to_instant();
match mode {
ByMode::Notify => {
for domain in &mut message.domains {
domain.notify.due = alimit;
domain.notify.due = alimit as u64;
}
}
ByMode::Return => {
for domain in &mut message.domains {
domain.expires = alimit;
domain.expires = alimit as u64;
}
}
ByMode::Default => (),
@@ -302,10 +294,10 @@ impl SMTP {
None
};
handle.block_on(self.queue.queue_message(
message,
handle.block_on(message.queue(
headers.as_deref(),
raw_message,
self,
&span,
));
}

26
crates/smtp/src/scripts/plugins/bayes.rs
View File

@@ -29,12 +29,12 @@ use nlp::{
tokenizers::osb::{OsbToken, OsbTokenizer},
};
use sieve::{runtime::Variable, FunctionMap};
use store::{write::key::KeySerializer, LookupKey, LookupStore, LookupValue, U64_LEN};
use store::{write::key::KeySerializer, LookupStore, U64_LEN};
use tokio::runtime::Handle;
use crate::config::scripts::SieveContext;
use super::{lookup::VariableExists, PluginContext};
use super::PluginContext;
pub fn register_train(plugin_id: u32, fnc_map: &mut FunctionMap<SieveContext>) {
fnc_map.set_external_function("bayes_train", plugin_id, 3);
@@ -110,14 +110,13 @@ fn train(ctx: PluginContext<'_>, is_train: bool) -> Variable {
for (hash, weights) in model.weights {
if handle
.block_on(
store.key_set(
store.counter_incr(
KeySerializer::new(U64_LEN)
.write(hash.h1)
.write(hash.h2)
.finalize(),
LookupValue::Counter {
num: weights.into(),
},
weights.into(),
None,
),
)
.is_err()
@@ -135,14 +134,13 @@ fn train(ctx: PluginContext<'_>, is_train: bool) -> Variable {
};
if handle
.block_on(
store.key_set(
store.counter_incr(
KeySerializer::new(U64_LEN)
.write(0u64)
.write(0u64)
.finalize(),
LookupValue::Counter {
num: weights.into(),
},
weights.into(),
None,
),
)
.is_err()
@@ -337,15 +335,15 @@ impl LookupOrInsert for BayesTokenCache {
) -> Option<Weights> {
if let Some(weights) = self.get(&hash) {
weights.unwrap_or_default().into()
} else if let Ok(result) = handle.block_on(
get_token.key_get::<VariableExists>(LookupKey::Counter(
} else if let Ok(num) = handle.block_on(
get_token.counter_get(
KeySerializer::new(U64_LEN)
.write(hash.h1)
.write(hash.h2)
.finalize(),
)),
),
) {
if let LookupValue::Counter { num } = result {
if num != 0 {
let weights = Weights::from(num);
self.insert_positive(hash, weights);
weights

60
crates/smtp/src/scripts/plugins/lookup.rs
View File

@@ -29,7 +29,7 @@ use std::{
use mail_auth::flate2;
use sieve::{runtime::Variable, FunctionMap};
use store::{Deserialize, LookupKey, LookupValue, Value};
use store::{Deserialize, Value};
use crate::{
config::scripts::{RemoteList, SieveContext},
@@ -72,10 +72,7 @@ pub fn exec(ctx: PluginContext<'_>) -> Variable {
if !item.is_empty()
&& ctx
.handle
.block_on(store.key_get::<VariableExists>(LookupKey::Key(
item.to_string().into_owned().into_bytes(),
)))
.map(|v| v != LookupValue::None)
.block_on(store.key_exists(item.to_string().into_owned().into_bytes()))
.unwrap_or(false)
{
return true.into();
@@ -85,10 +82,7 @@ pub fn exec(ctx: PluginContext<'_>) -> Variable {
}
v if !v.is_empty() => ctx
.handle
.block_on(store.key_get::<VariableExists>(LookupKey::Key(
v.to_string().into_owned().into_bytes(),
)))
.map(|v| v != LookupValue::None)
.block_on(store.key_exists(v.to_string().into_owned().into_bytes()))
.unwrap_or(false),
_ => false,
}
@@ -113,14 +107,13 @@ pub fn exec_get(ctx: PluginContext<'_>) -> Variable {
if let Some(store) = store {
ctx.handle
.block_on(store.key_get::<VariableWrapper>(LookupKey::Key(
ctx.arguments[1].to_string().into_owned().into_bytes(),
)))
.map(|v| match v {
LookupValue::Value { value, .. } => value.into_inner(),
LookupValue::Counter { num } => num.into(),
LookupValue::None => Variable::default(),
})
.block_on(
store.key_get::<VariableWrapper>(
ctx.arguments[1].to_string().into_owned().into_bytes(),
),
)
.unwrap_or_default()
.map(|v| v.into_inner())
.unwrap_or_default()
} else {
tracing::warn!(
@@ -142,22 +135,20 @@ pub fn exec_set(ctx: PluginContext<'_>) -> Variable {
if let Some(store) = store {
let expires = match &ctx.arguments[3] {
Variable::Integer(v) => *v as u64,
Variable::Float(v) => *v as u64,
_ => 0,
Variable::Integer(v) => Some(*v as u64),
Variable::Float(v) => Some(*v as u64),
_ => None,
};
ctx.handle
.block_on(store.key_set(
ctx.arguments[1].to_string().into_owned().into_bytes(),
LookupValue::Value {
value: if !ctx.arguments[2].is_empty() {
bincode::serialize(&ctx.arguments[2]).unwrap_or_default()
} else {
vec![]
},
expires,
if !ctx.arguments[2].is_empty() {
bincode::serialize(&ctx.arguments[2]).unwrap_or_default()
} else {
vec![]
},
expires,
))
.is_ok()
.into()
@@ -426,9 +417,6 @@ pub fn exec_local_domain(ctx: PluginContext<'_>) -> Variable {
#[derive(Debug, PartialEq, Eq)]
pub struct VariableWrapper(Variable);
#[derive(Debug, PartialEq, Eq)]
pub struct VariableExists;
impl Deserialize for VariableWrapper {
fn deserialize(bytes: &[u8]) -> store::Result<Self> {
Ok(VariableWrapper(
@@ -439,9 +427,9 @@ impl Deserialize for VariableWrapper {
}
}
impl Deserialize for VariableExists {
fn deserialize(_: &[u8]) -> store::Result<Self> {
Ok(VariableExists)
impl From<i64> for VariableWrapper {
fn from(value: i64) -> Self {
VariableWrapper(value.into())
}
}
@@ -451,12 +439,6 @@ impl VariableWrapper {
}
}
impl From<Value<'static>> for VariableExists {
fn from(_: Value<'static>) -> Self {
VariableExists
}
}
impl From<Value<'static>> for VariableWrapper {
fn from(value: Value<'static>) -> Self {
VariableWrapper(into_sieve_value(value))