Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 jcoffey-dev
|
8a8f48c944 |
@@ -3,28 +3,11 @@
|
||||
# whether a person pushed it or weekly-release.yml created it through the
|
||||
# releases API.
|
||||
#
|
||||
# The image is multi-arch (linux/amd64, linux/arm64), built by two jobs on
|
||||
# the image-build runner rather than one buildx run for both. The Dockerfile's
|
||||
# builder stage runs on the build platform and cross-compiles with an aarch64
|
||||
# linker, so only the small final stage (apt, setcap) goes through QEMU for
|
||||
# arm64 -- but two release builds (LTO, one codegen unit) side by side on one
|
||||
# machine each take twice as long. Production runs amd64, so amd64 goes first
|
||||
# and on its own:
|
||||
# * publish-amd64 pushes :<version>-amd64 and :<version>, a plain amd64
|
||||
# image, as soon as its build is done. A deploy can start from it.
|
||||
# * publish-arm64 then builds arm64, pushes :<version>-arm64, and replaces
|
||||
# :<version> with the two-platform index. :latest moves only here, so it
|
||||
# never names an image without arm64.
|
||||
#
|
||||
# Both jobs use one BuildKit builder, `gitea-builder`, whose container
|
||||
# (buildx_buildkit_gitea-builder0) and state volume stay on the runner's host
|
||||
# between jobs: a job container's `buildx create` finds the existing container
|
||||
# and reuses it and its cache. The dependency build (`cargo chef cook`) is
|
||||
# keyed on the recipe, which only a dependency change alters, so a release
|
||||
# normally compiles just the workspace. Removing that container or its volume
|
||||
# costs the next release a cold build, nothing more. The planner and dependency
|
||||
# layers for the build platform are shared, so arm64 also reuses what amd64
|
||||
# just did where it can.
|
||||
# The image is multi-arch (linux/amd64, linux/arm64) as before, but built in
|
||||
# one buildx run on host1 instead of one native runner per architecture: the
|
||||
# Dockerfile's builder stage runs on the build platform and cross-compiles
|
||||
# with an aarch64 linker, so only the small final stage (apt, setcap) goes
|
||||
# through QEMU for arm64. No digest-joining job is needed.
|
||||
#
|
||||
# Two guards before anything is pushed:
|
||||
# * the tag must be v<brand_version!>. The version is a string in
|
||||
@@ -79,7 +62,7 @@ jobs:
|
||||
echo "version=$V" >> "$GITHUB_OUTPUT"
|
||||
echo "version $V"
|
||||
|
||||
publish-amd64:
|
||||
publish:
|
||||
needs: [version]
|
||||
runs-on: docker
|
||||
container:
|
||||
@@ -98,15 +81,16 @@ jobs:
|
||||
test -n "$REGISTRY" && test -n "$VERSION"
|
||||
test -n "$PACKAGE_TOKEN" || { echo "PACKAGE_TOKEN secret is not set on this repository" >&2; exit 1; }
|
||||
echo "$PACKAGE_TOKEN" | docker login -u jcoffey-dev --password-stdin "$REGISTRY"
|
||||
docker run --privileged --rm tonistiigi/binfmt --install arm64
|
||||
docker buildx create --use --name gitea-builder --driver docker-container || docker buildx use gitea-builder
|
||||
# Attestations off, as before: they add manifests of their own, and the
|
||||
# index should hold the two images and nothing else.
|
||||
# Attestations off, as before: they add manifests of their own to the
|
||||
# index, and the index should hold the two images and nothing else.
|
||||
- run: |
|
||||
docker buildx build \
|
||||
--platform linux/amd64 \
|
||||
--platform linux/amd64,linux/arm64 \
|
||||
--provenance=false --sbom=false \
|
||||
--tag "$IMAGE:$VERSION-amd64" \
|
||||
--tag "$IMAGE:$VERSION" \
|
||||
--tag "$IMAGE:latest" \
|
||||
--push .
|
||||
docker buildx imagetools inspect "$IMAGE:$VERSION"
|
||||
# Gitea keeps a container package on its owner; linking it shows it on
|
||||
@@ -119,47 +103,11 @@ jobs:
|
||||
- if: always()
|
||||
run: docker logout "$REGISTRY" || true
|
||||
|
||||
publish-arm64:
|
||||
needs: [version, publish-amd64]
|
||||
runs-on: docker
|
||||
container:
|
||||
image: docker:28-cli@sha256:625d9431a9f54c5a2bc90f24f0e1c3d55b1349fd857dd85035f98c2c9acbdd4d # 28-cli
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
env:
|
||||
DOCKER_BUILDKIT: "1"
|
||||
REGISTRY: ${{ vars.REGISTRY }}
|
||||
IMAGE: ${{ vars.REGISTRY }}/${{ github.repository }}
|
||||
VERSION: ${{ needs.version.outputs.version }}
|
||||
PACKAGE_TOKEN: ${{ secrets.PACKAGE_TOKEN }}
|
||||
steps:
|
||||
- uses: coffey-labs/actions/checkout@fab0c4d45e0162963965f1555df27b7bed5e20ec
|
||||
- run: |
|
||||
echo "$PACKAGE_TOKEN" | docker login -u jcoffey-dev --password-stdin "$REGISTRY"
|
||||
docker run --privileged --rm tonistiigi/binfmt --install arm64
|
||||
docker buildx create --use --name gitea-builder --driver docker-container || docker buildx use gitea-builder
|
||||
# The index is built from the two per-architecture tags rather than from
|
||||
# :<version>, which by now is the amd64 image and would be read as such.
|
||||
- run: |
|
||||
docker buildx build \
|
||||
--platform linux/arm64 \
|
||||
--provenance=false --sbom=false \
|
||||
--tag "$IMAGE:$VERSION-arm64" \
|
||||
--push .
|
||||
docker buildx imagetools create \
|
||||
--tag "$IMAGE:$VERSION" \
|
||||
--tag "$IMAGE:latest" \
|
||||
"$IMAGE:$VERSION-amd64" "$IMAGE:$VERSION-arm64"
|
||||
docker buildx imagetools inspect "$IMAGE:$VERSION"
|
||||
- if: always()
|
||||
run: docker logout "$REGISTRY" || true
|
||||
|
||||
# The weekly release creates its Release (and so the tag) first; a tag
|
||||
# pushed by hand has none. Either way the tag ends up with exactly one
|
||||
# Release, created once the amd64 image exists so its pull instructions
|
||||
# work; arm64 and the binaries follow.
|
||||
# Release, created after the image exists so its pull instructions work.
|
||||
release:
|
||||
needs: [version, publish-amd64]
|
||||
needs: [version, publish]
|
||||
runs-on: light
|
||||
container:
|
||||
image: python:3.13-slim@sha256:8d9d0b8bcf6506481eae4907c18f5e3e7902e629f5f6d684f9e7c32e85e3ddf0 # 3.13-slim
|
||||
@@ -183,9 +131,7 @@ jobs:
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code != 404: raise
|
||||
image = f"{os.environ['REGISTRY']}/{os.environ['REPO']}:{version}"
|
||||
body = (f"Container image: `{image}` (linux/amd64, linux/arm64); also `:latest`. "
|
||||
"amd64 is published first; arm64 is added to the same tag when its build "
|
||||
"finishes, and `:latest` moves then.\n\n"
|
||||
body = (f"Container image: `{image}` (linux/amd64, linux/arm64); also `:latest`.\n\n"
|
||||
"Binaries for a host install are attached: `inbuxa-linux-amd64.tar.gz` and "
|
||||
"`inbuxa-linux-arm64.tar.gz`, with `SHA256SUMS`. Each is the binary out of this "
|
||||
"release's image for that architecture, so it is the same build. The image "
|
||||
@@ -208,7 +154,7 @@ jobs:
|
||||
# `docker create` does not start anything, so pulling an arm64 image on an
|
||||
# amd64 runner and copying a file out of it needs no emulation.
|
||||
binaries:
|
||||
needs: [version, publish-arm64, release]
|
||||
needs: [version, publish, release]
|
||||
runs-on: docker
|
||||
container:
|
||||
image: docker:28-cli@sha256:625d9431a9f54c5a2bc90f24f0e1c3d55b1349fd857dd85035f98c2c9acbdd4d # 28-cli
|
||||
|
||||
@@ -335,57 +335,3 @@ impl EmailPush {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// inbuxa: the task locks this node holds, so a graceful stop can hand them
|
||||
/// back instead of leaving the tasks blocked until the locks expire.
|
||||
pub struct TaskLocks {
|
||||
held: parking_lot::Mutex<ahash::AHashSet<u64>>,
|
||||
stopping: AtomicBool,
|
||||
expiry: std::sync::atomic::AtomicU64,
|
||||
}
|
||||
|
||||
impl TaskLocks {
|
||||
/// How long a task lock lasts, in seconds, unless it is released first.
|
||||
pub const DEFAULT_EXPIRY: u64 = 60 * 60;
|
||||
|
||||
pub fn is_stopping(&self) -> bool {
|
||||
self.stopping.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Stops new claims and returns the ids of every lock still held.
|
||||
pub fn stop(&self) -> Vec<u64> {
|
||||
self.stopping.store(true, Ordering::Release);
|
||||
self.held.lock().drain().collect()
|
||||
}
|
||||
|
||||
pub fn insert(&self, id: u64) {
|
||||
self.held.lock().insert(id);
|
||||
}
|
||||
|
||||
pub fn remove(&self, id: u64) {
|
||||
self.held.lock().remove(&id);
|
||||
}
|
||||
|
||||
pub fn held(&self) -> usize {
|
||||
self.held.lock().len()
|
||||
}
|
||||
|
||||
pub fn expiry(&self) -> u64 {
|
||||
self.expiry.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// Changes the lock lifetime; the tests shorten it.
|
||||
pub fn set_expiry(&self, seconds: u64) {
|
||||
self.expiry.store(seconds.max(1), Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for TaskLocks {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
held: Default::default(),
|
||||
stopping: AtomicBool::new(false),
|
||||
expiry: std::sync::atomic::AtomicU64::new(Self::DEFAULT_EXPIRY),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -279,8 +279,6 @@ pub struct HttpAuthCache {
|
||||
pub struct Ipc {
|
||||
pub push_tx: mpsc::Sender<PushEvent>,
|
||||
pub task_tx: Arc<Notify>,
|
||||
// inbuxa: task locks held by this node, released on a graceful stop
|
||||
pub task_locks: Arc<crate::ipc::TaskLocks>,
|
||||
pub queue_tx: mpsc::Sender<QueueEvent>,
|
||||
pub report_tx: mpsc::Sender<ReportingEvent>,
|
||||
pub broadcast_tx: Option<mpsc::Sender<BroadcastEvent>>,
|
||||
|
||||
@@ -297,7 +297,6 @@ pub fn build_ipc(has_pubsub: bool) -> (Ipc, IpcReceivers) {
|
||||
report_tx,
|
||||
broadcast_tx: has_pubsub.then_some(broadcast_tx),
|
||||
task_tx: Arc::new(Notify::new()),
|
||||
task_locks: Arc::new(crate::ipc::TaskLocks::default()),
|
||||
train_task_controller: Arc::new(TrainTaskController::default()),
|
||||
},
|
||||
IpcReceivers {
|
||||
|
||||
@@ -109,10 +109,6 @@ async fn main() -> std::io::Result<()> {
|
||||
// Wait for shutdown signal
|
||||
wait_for_shutdown().await;
|
||||
|
||||
// inbuxa: hand back the task locks this node holds, so other nodes can
|
||||
// run those tasks now rather than when the locks expire
|
||||
services::task_manager::lock::release_task_locks(&inner.build_server()).await;
|
||||
|
||||
// Shutdown collector
|
||||
Collector::shutdown();
|
||||
|
||||
|
||||
@@ -91,15 +91,7 @@ impl SearchIndexTask for Server {
|
||||
build_contact_document(self, account_id, document_id).await
|
||||
}
|
||||
IndexDocumentType::File => {
|
||||
// File indexing not implemented yet. inbuxa: still
|
||||
// one result per task: update_tasks pairs them by
|
||||
// position, and a missing one shifts every result
|
||||
// after it onto the wrong task
|
||||
results.push(IndexTaskResult {
|
||||
task_type: TaskType::Insert,
|
||||
index: task.document_type,
|
||||
result: TaskResult::Ignored,
|
||||
});
|
||||
// File indexing not implemented yet
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -2,8 +2,6 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::task_manager::*;
|
||||
@@ -15,21 +13,13 @@ pub trait TaskLockManager: Sync + Send {
|
||||
|
||||
impl TaskLockManager for Server {
|
||||
async fn try_lock_task(&self, id: u64) -> bool {
|
||||
// inbuxa: a node that is stopping claims nothing new
|
||||
let locks = &self.inner.ipc.task_locks;
|
||||
if locks.is_stopping() {
|
||||
return false;
|
||||
}
|
||||
|
||||
match self
|
||||
.in_memory_store()
|
||||
.try_lock(KV_LOCK_TASK, &id.to_be_bytes(), locks.expiry())
|
||||
.try_lock(KV_LOCK_TASK, &id.to_be_bytes(), DEFAULT_LOCK_EXPIRY)
|
||||
.await
|
||||
{
|
||||
Ok(result) => {
|
||||
if result {
|
||||
locks.insert(id);
|
||||
} else {
|
||||
if !result {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskLocked),
|
||||
Id = id,
|
||||
@@ -58,27 +48,5 @@ impl TaskLockManager for Server {
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
self.inner.ipc.task_locks.remove(id);
|
||||
}
|
||||
}
|
||||
|
||||
/// inbuxa: on a graceful stop, stops claiming tasks and releases every task
|
||||
/// lock this node holds, so the rest of the cluster can pick the tasks up at
|
||||
/// once instead of after the lock expires. Returns how many were released.
|
||||
pub async fn release_task_locks(server: &Server) -> usize {
|
||||
let ids = server.inner.ipc.task_locks.stop();
|
||||
for id in &ids {
|
||||
if let Err(err) = server
|
||||
.in_memory_store()
|
||||
.remove_lock(KV_LOCK_TASK, &id.to_be_bytes())
|
||||
.await
|
||||
{
|
||||
trc::error!(
|
||||
err.details("Failed to release task lock on shutdown")
|
||||
.ctx(trc::Key::Id, *id)
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
}
|
||||
ids.len()
|
||||
}
|
||||
|
||||
@@ -2,8 +2,6 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::task_manager::acme::AcmeTask;
|
||||
@@ -20,7 +18,7 @@ use crate::task_manager::report::{self, SubmitReportTask};
|
||||
use crate::task_manager::restore_item::RestoreItemTask;
|
||||
use crate::task_manager::spam_classifier::SpamFilterMaintenanceTask;
|
||||
use crate::task_manager::{
|
||||
CLAIM_RECHECK_INTERVAL, Locked, QUEUE_REFRESH_INTERVAL, TaskDetails, TaskFailureType, TaskInfo,
|
||||
DEFAULT_LOCK_EXPIRY, Locked, QUEUE_REFRESH_INTERVAL, TaskDetails, TaskFailureType, TaskInfo,
|
||||
TaskJob, TaskManagerIpc, TaskResult,
|
||||
};
|
||||
use common::BuildServer;
|
||||
@@ -126,47 +124,72 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
|
||||
let server = inner.build_server();
|
||||
let batch_size = server.core.email.index_batch_size;
|
||||
let mut batch = Vec::with_capacity(batch_size);
|
||||
if let Some(task) = fetch_task(&server, job).await {
|
||||
batch.push(task);
|
||||
match server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id: job.id },
|
||||
)))
|
||||
.await
|
||||
{
|
||||
Ok(Some(task)) => {
|
||||
batch.push(TaskDetails { task, info: job });
|
||||
}
|
||||
Ok(None) => {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskIgnored),
|
||||
Id = job.id,
|
||||
Reason = "Task not found in store, likely already processed.",
|
||||
);
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(job.id)
|
||||
.details("Failed to retrieve task details.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
while batch.len() < batch_size {
|
||||
match rx.try_recv() {
|
||||
Ok(job) => {
|
||||
if let Some(task) = fetch_task(&server, job).await {
|
||||
batch.push(task);
|
||||
match server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id: job.id },
|
||||
)))
|
||||
.await
|
||||
{
|
||||
Ok(Some(task)) => {
|
||||
batch.push(TaskDetails { task, info: job });
|
||||
}
|
||||
Ok(None) => {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskIgnored),
|
||||
Id = job.id,
|
||||
Reason = "Task not found in store, likely already processed.",
|
||||
);
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(job.id)
|
||||
.details("Failed to retrieve task details.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
|
||||
// Dispatch. inbuxa: on a task of its own, so a panic
|
||||
// releases the batch's locks and leaves this worker
|
||||
// running; a dead worker would keep claiming tasks it
|
||||
// can never run
|
||||
// Dispatch
|
||||
let mut refresh_queue = false;
|
||||
let ids = batch.iter().map(|task| task.info.id).collect::<Vec<_>>();
|
||||
let run = {
|
||||
let server = server.clone();
|
||||
tokio::spawn(async move {
|
||||
let results = server.index(&batch).await;
|
||||
(batch, results)
|
||||
})
|
||||
};
|
||||
match run.await {
|
||||
Ok((mut batch, results)) => {
|
||||
let results = results.into_iter().map(|r| {
|
||||
refresh_queue |= r.result.is_retry();
|
||||
r.result
|
||||
});
|
||||
update_tasks(&server, &mut batch, results).await;
|
||||
}
|
||||
Err(err) => {
|
||||
worker_failed(&server, &ids, err).await;
|
||||
refresh_queue = true;
|
||||
}
|
||||
}
|
||||
let results = server.index(&batch).await.into_iter().map(|r| {
|
||||
refresh_queue |= r.result.is_retry();
|
||||
r.result
|
||||
});
|
||||
update_tasks(&server, &mut batch, results).await;
|
||||
|
||||
if refresh_queue || rx.is_empty() {
|
||||
server.notify_task_queue();
|
||||
@@ -180,31 +203,83 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
|
||||
let server = inner.build_server();
|
||||
let mut refresh_queue = false;
|
||||
|
||||
if let Some(TaskDetails { task, info }) = fetch_task(&server, job).await {
|
||||
// inbuxa: on a task of its own, as above
|
||||
let run = {
|
||||
let server = server.clone();
|
||||
let server_instance = server_instance.clone();
|
||||
tokio::spawn(async move {
|
||||
let result = run_task(&server, &task, server_instance).await;
|
||||
(task, result)
|
||||
})
|
||||
};
|
||||
match run.await {
|
||||
Ok((task, result)) => {
|
||||
refresh_queue = result.is_retry();
|
||||
match server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id: job.id },
|
||||
)))
|
||||
.await
|
||||
{
|
||||
Ok(Some(task)) => {
|
||||
let result = match &task {
|
||||
Task::CalendarAlarmEmail(task) => {
|
||||
server.send_email_alarm(task, server_instance.clone()).await
|
||||
}
|
||||
Task::CalendarAlarmNotification(task) => {
|
||||
server.send_display_alarm(task).await
|
||||
}
|
||||
Task::CalendarItipMessage(task) => {
|
||||
server.send_imip(task, server_instance.clone()).await
|
||||
}
|
||||
Task::MergeThreads(task) => server.merge_threads(task).await,
|
||||
Task::DmarcReport(task) => {
|
||||
server
|
||||
.submit_report(report::ReportId::Dmarc(task.report_id.id()))
|
||||
.await
|
||||
}
|
||||
Task::TlsReport(task) => {
|
||||
server
|
||||
.submit_report(report::ReportId::Tls(task.report_id.id()))
|
||||
.await
|
||||
}
|
||||
Task::RestoreArchivedItem(task) => server.restore_item(task).await,
|
||||
Task::DestroyAccount(task) => server.destroy_account(task).await,
|
||||
Task::AccountMaintenance(task) => {
|
||||
server.account_maintenance(task).await
|
||||
}
|
||||
Task::TenantMaintenance(task) => {
|
||||
server.tenant_maintenance(task).await
|
||||
}
|
||||
Task::StoreMaintenance(task) => {
|
||||
server.store_maintenance(task).await
|
||||
}
|
||||
Task::SpamFilterMaintenance(task) => {
|
||||
Box::pin(server.spam_filter_maintenance(task)).await
|
||||
}
|
||||
Task::AcmeRenewal(task) => server.acme_management(task).await,
|
||||
Task::DkimManagement(task_dkim_rotation) => {
|
||||
server.dkim_management(task_dkim_rotation).await
|
||||
}
|
||||
Task::DnsManagement(task_dns_management) => {
|
||||
server.dns_management(task_dns_management).await
|
||||
}
|
||||
Task::IndexDocument(_)
|
||||
| Task::UnindexDocument(_)
|
||||
| Task::IndexTrace(_) => unreachable!(),
|
||||
};
|
||||
|
||||
update_tasks(
|
||||
&server,
|
||||
&mut [TaskDetails { task, info }],
|
||||
vec![result],
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Err(err) => {
|
||||
worker_failed(&server, &[info.id], err).await;
|
||||
refresh_queue = true;
|
||||
}
|
||||
refresh_queue = result.is_retry();
|
||||
|
||||
update_tasks(
|
||||
&server,
|
||||
&mut [TaskDetails { task, info: job }],
|
||||
vec![result],
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Ok(None) => {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskIgnored),
|
||||
Id = job.id,
|
||||
Reason = "Task not found in store, likely already processed.",
|
||||
);
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(job.id)
|
||||
.details("Failed to retrieve task details.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -243,13 +318,6 @@ pub(crate) trait TaskQueueManager: Sync + Send {
|
||||
|
||||
impl TaskQueueManager for Server {
|
||||
async fn process_tasks(&self, ipc: &mut TaskManagerIpc) -> Duration {
|
||||
// inbuxa: a node that is stopping has released its locks and claims
|
||||
// nothing new
|
||||
let task_locks = &self.inner.ipc.task_locks;
|
||||
if task_locks.is_stopping() {
|
||||
return Duration::from_secs(QUEUE_REFRESH_INTERVAL);
|
||||
}
|
||||
let lock_expiry = task_locks.expiry();
|
||||
let now_timestamp = now();
|
||||
let from_key = ValueKey::<ValueClass> {
|
||||
account_id: 0,
|
||||
@@ -325,7 +393,9 @@ impl TaskQueueManager for Server {
|
||||
let locked = entry.get_mut();
|
||||
if locked.expires <= now || locked.due < task_due {
|
||||
locked.expires = Instant::now()
|
||||
+ std::time::Duration::from_secs(lock_expiry + 1);
|
||||
+ std::time::Duration::from_secs(
|
||||
DEFAULT_LOCK_EXPIRY + 1,
|
||||
);
|
||||
locked.due = task_due;
|
||||
tasks.push((
|
||||
TaskJob {
|
||||
@@ -341,7 +411,9 @@ impl TaskQueueManager for Server {
|
||||
Entry::Vacant(entry) => {
|
||||
entry.insert(Locked {
|
||||
expires: Instant::now()
|
||||
+ std::time::Duration::from_secs(lock_expiry + 1),
|
||||
+ std::time::Duration::from_secs(
|
||||
DEFAULT_LOCK_EXPIRY + 1,
|
||||
),
|
||||
due: task_due,
|
||||
revision: ipc.revision,
|
||||
});
|
||||
@@ -392,26 +464,12 @@ impl TaskQueueManager for Server {
|
||||
let tx = &ipc.txs[task_type_idx as usize];
|
||||
|
||||
if tx.capacity() > 0 {
|
||||
let id = task_job.id;
|
||||
if !self.try_lock_task(id).await {
|
||||
// inbuxa: another node holds the task. Look again after a
|
||||
// short while rather than a full lock lifetime from now:
|
||||
// the holder may have claimed it after this scan began,
|
||||
// or run on a clock ahead of this one, and waiting the
|
||||
// whole lifetime again would leave the task stuck for
|
||||
// another hour past its lock if that holder died
|
||||
if let Some(locked) = ipc.locked.get_mut(&id) {
|
||||
locked.expires =
|
||||
Instant::now() + Duration::from_secs(claim_recheck_interval(lock_expiry));
|
||||
}
|
||||
} else if tx.send(task_job).await.is_err() {
|
||||
if self.try_lock_task(task_job.id).await && tx.send(task_job).await.is_err() {
|
||||
trc::event!(
|
||||
Server(trc::ServerEvent::ThreadError),
|
||||
Details = "Error sending task.",
|
||||
CausedBy = trc::location!()
|
||||
);
|
||||
// inbuxa: nothing will run it here, so don't hold it
|
||||
self.remove_index_lock(id).await;
|
||||
}
|
||||
} else {
|
||||
// If the channel is full, release the lock so it can be picked up in the next iteration
|
||||
@@ -423,117 +481,9 @@ impl TaskQueueManager for Server {
|
||||
let now = Instant::now();
|
||||
ipc.locked
|
||||
.retain(|_, locked| locked.expires > now && locked.revision == ipc.revision);
|
||||
let sleep_for = Duration::from_secs(next_event.map_or(QUEUE_REFRESH_INTERVAL, |timestamp| {
|
||||
Duration::from_secs(next_event.map_or(QUEUE_REFRESH_INTERVAL, |timestamp| {
|
||||
timestamp.saturating_sub(store::write::now())
|
||||
}));
|
||||
|
||||
// inbuxa: wake up when a claim held elsewhere is due to be tried
|
||||
// again, rather than only on the next task or refresh
|
||||
ipc.locked
|
||||
.values()
|
||||
.map(|locked| locked.expires.saturating_duration_since(now))
|
||||
.min()
|
||||
.map_or(sleep_for, |recheck| sleep_for.min(recheck.max(Duration::from_secs(1))))
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_task(
|
||||
server: &Server,
|
||||
task: &Task,
|
||||
server_instance: Arc<ServerInstance>,
|
||||
) -> TaskResult {
|
||||
match task {
|
||||
Task::CalendarAlarmEmail(task) => {
|
||||
server.send_email_alarm(task, server_instance.clone()).await
|
||||
}
|
||||
Task::CalendarAlarmNotification(task) => {
|
||||
server.send_display_alarm(task).await
|
||||
}
|
||||
Task::CalendarItipMessage(task) => {
|
||||
server.send_imip(task, server_instance.clone()).await
|
||||
}
|
||||
Task::MergeThreads(task) => server.merge_threads(task).await,
|
||||
Task::DmarcReport(task) => {
|
||||
server
|
||||
.submit_report(report::ReportId::Dmarc(task.report_id.id()))
|
||||
.await
|
||||
}
|
||||
Task::TlsReport(task) => {
|
||||
server
|
||||
.submit_report(report::ReportId::Tls(task.report_id.id()))
|
||||
.await
|
||||
}
|
||||
Task::RestoreArchivedItem(task) => server.restore_item(task).await,
|
||||
Task::DestroyAccount(task) => server.destroy_account(task).await,
|
||||
Task::AccountMaintenance(task) => {
|
||||
server.account_maintenance(task).await
|
||||
}
|
||||
Task::TenantMaintenance(task) => {
|
||||
server.tenant_maintenance(task).await
|
||||
}
|
||||
Task::StoreMaintenance(task) => {
|
||||
server.store_maintenance(task).await
|
||||
}
|
||||
Task::SpamFilterMaintenance(task) => {
|
||||
Box::pin(server.spam_filter_maintenance(task)).await
|
||||
}
|
||||
Task::AcmeRenewal(task) => server.acme_management(task).await,
|
||||
Task::DkimManagement(task_dkim_rotation) => {
|
||||
server.dkim_management(task_dkim_rotation).await
|
||||
}
|
||||
Task::DnsManagement(task_dns_management) => {
|
||||
server.dns_management(task_dns_management).await
|
||||
}
|
||||
Task::IndexDocument(_)
|
||||
| Task::UnindexDocument(_)
|
||||
| Task::IndexTrace(_) => unreachable!(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads a claimed task. When it is gone or can't be read, the claim is
|
||||
/// released: inbuxa: holding it would block the task, everywhere, until
|
||||
/// the lock expired.
|
||||
async fn fetch_task(server: &Server, job: TaskJob) -> Option<TaskDetails> {
|
||||
match server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Task {
|
||||
id: job.id,
|
||||
})))
|
||||
.await
|
||||
{
|
||||
Ok(Some(task)) => Some(TaskDetails { task, info: job }),
|
||||
Ok(None) => {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskIgnored),
|
||||
Id = job.id,
|
||||
Reason = "Task not found in store, likely already processed.",
|
||||
);
|
||||
server.remove_index_lock(job.id).await;
|
||||
None
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(job.id)
|
||||
.details("Failed to retrieve task details.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
server.remove_index_lock(job.id).await;
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// inbuxa: a task panicked: its locks are released so it runs again, here or
|
||||
/// on another node, and the worker carries on.
|
||||
async fn worker_failed(server: &Server, ids: &[u64], err: tokio::task::JoinError) {
|
||||
trc::event!(
|
||||
Server(trc::ServerEvent::ThreadError),
|
||||
Details = "Task worker failed",
|
||||
Reason = err.to_string(),
|
||||
CausedBy = trc::location!()
|
||||
);
|
||||
for id in ids {
|
||||
server.remove_index_lock(*id).await;
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -664,13 +614,6 @@ async fn update_tasks(
|
||||
}
|
||||
}
|
||||
|
||||
/// inbuxa: how long to wait before trying again to claim a task another node
|
||||
/// holds: a twelfth of the lock lifetime, so five minutes for the one-hour
|
||||
/// lock, never more than that and never under a second.
|
||||
pub(crate) fn claim_recheck_interval(lock_expiry: u64) -> u64 {
|
||||
(lock_expiry / 12).clamp(1, CLAIM_RECHECK_INTERVAL)
|
||||
}
|
||||
|
||||
pub fn perpetual_retry_time(typ: TaskType, attempt: u64) -> Option<u64> {
|
||||
matches!(
|
||||
typ,
|
||||
|
||||
@@ -35,9 +35,7 @@ pub mod scheduler;
|
||||
pub mod spam_classifier;
|
||||
|
||||
const QUEUE_REFRESH_INTERVAL: u64 = 60 * 5; // 5 minutes
|
||||
// inbuxa: the lock lifetime (one hour) lives in common::ipc::TaskLocks, per
|
||||
// server, so a graceful stop can release the locks and the tests can shorten it
|
||||
const CLAIM_RECHECK_INTERVAL: u64 = 60 * 5; // 5 minutes
|
||||
const DEFAULT_LOCK_EXPIRY: u64 = 60 * 60; // 1 hour
|
||||
|
||||
pub(crate) struct TaskManagerIpc {
|
||||
txs: [mpsc::Sender<TaskJob>; TaskType::COUNT],
|
||||
|
||||
@@ -2,8 +2,6 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
@@ -21,7 +19,7 @@ use crate::{
|
||||
write::SearchIndex,
|
||||
};
|
||||
use mysql_async::{IsolationLevel, TxOpts, Value, prelude::Queryable};
|
||||
use nlp::{language::Language, tokenizers::word::WordTokenizer};
|
||||
use nlp::tokenizers::word::WordTokenizer;
|
||||
use std::fmt::Write;
|
||||
|
||||
impl MysqlStore {
|
||||
@@ -148,20 +146,6 @@ impl MysqlStore {
|
||||
}
|
||||
}
|
||||
|
||||
// inbuxa: InnoDB's default full-text stopword list
|
||||
// (INFORMATION_SCHEMA.INNODB_FT_DEFAULT_STOPWORD) and innodb_ft_min_token_size
|
||||
// default; words outside these are not in a FULLTEXT index.
|
||||
const FT_STOPWORDS: &[&str] = &[
|
||||
"a", "about", "an", "are", "as", "at", "be", "by", "com", "de", "en", "for", "from", "how",
|
||||
"i", "in", "is", "it", "la", "of", "on", "or", "that", "the", "this", "to", "was", "what",
|
||||
"when", "where", "who", "will", "with", "und", "www",
|
||||
];
|
||||
const FT_MIN_TOKEN_SIZE: usize = 3;
|
||||
|
||||
fn is_ft_indexed(word: &str) -> bool {
|
||||
word.chars().count() >= FT_MIN_TOKEN_SIZE && !FT_STOPWORDS.contains(&word)
|
||||
}
|
||||
|
||||
fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
|
||||
if filters.is_empty() {
|
||||
return Vec::new();
|
||||
@@ -187,77 +171,30 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
|
||||
|
||||
if field.is_text() && matches!(op, SearchOperator::Equal | SearchOperator::Contains)
|
||||
{
|
||||
let (value, mode, unindexed) = match (value, op) {
|
||||
(SearchValue::Text { value, .. }, SearchOperator::Equal) => (
|
||||
Value::Bytes(format!("{value:?}").into_bytes()),
|
||||
"BOOLEAN",
|
||||
Vec::new(),
|
||||
),
|
||||
(SearchValue::Text { value, language }, ..) => {
|
||||
let (value, mode) = match (value, op) {
|
||||
(SearchValue::Text { value, .. }, SearchOperator::Equal) => {
|
||||
(Value::Bytes(format!("{value:?}").into_bytes()), "BOOLEAN")
|
||||
}
|
||||
(SearchValue::Text { value, .. }, ..) => {
|
||||
let mut text_query = String::with_capacity(value.len() + 1);
|
||||
let mut unindexed = Vec::new();
|
||||
|
||||
for item in WordTokenizer::new(value, MAX_TOKEN_LENGTH) {
|
||||
// inbuxa: InnoDB never indexes stopwords ("com",
|
||||
// "de", "www", ...) or words under
|
||||
// innodb_ft_min_token_size, and a required
|
||||
// (+word) term it has not indexed matches no row,
|
||||
// so "example.com" or "[email protected]" found
|
||||
// nothing. Such words are matched with a
|
||||
// word-boundary REGEXP instead.
|
||||
if is_ft_indexed(&item.word) {
|
||||
if !text_query.is_empty() {
|
||||
text_query.push(' ');
|
||||
}
|
||||
text_query.push('+');
|
||||
text_query.push_str(&item.word);
|
||||
} else {
|
||||
unindexed.push(item.word);
|
||||
if !text_query.is_empty() {
|
||||
text_query.push(' ');
|
||||
}
|
||||
text_query.push('+');
|
||||
text_query.push_str(&item.word);
|
||||
}
|
||||
|
||||
// For language text (bodies, subjects) the unindexed
|
||||
// words are noise words and only checked when nothing
|
||||
// else is left to match; keyword text (addresses,
|
||||
// contact fields) checks every word, as the other
|
||||
// backends do.
|
||||
if !text_query.is_empty() && !matches!(language, Language::None) {
|
||||
unindexed.clear();
|
||||
}
|
||||
|
||||
(Value::Bytes(text_query.into_bytes()), "BOOLEAN", unindexed)
|
||||
(Value::Bytes(text_query.into_bytes()), "BOOLEAN")
|
||||
}
|
||||
_ => {
|
||||
debug_assert!(false, "Invalid search value for text field");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
if unindexed.is_empty() {
|
||||
let _ =
|
||||
write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column());
|
||||
values.push(value);
|
||||
} else {
|
||||
query.push('(');
|
||||
let is_empty = matches!(&value, Value::Bytes(v) if v.is_empty());
|
||||
if !is_empty {
|
||||
let _ = write!(
|
||||
query,
|
||||
"MATCH({}) AGAINST(? IN {mode} MODE) AND ",
|
||||
field.column()
|
||||
);
|
||||
values.push(value);
|
||||
}
|
||||
for (i, word) in unindexed.iter().enumerate() {
|
||||
if i > 0 {
|
||||
query.push_str(" AND ");
|
||||
}
|
||||
let _ = write!(query, "{} REGEXP ?", field.column());
|
||||
values.push(Value::Bytes(
|
||||
format!("(^|[^[:alnum:]]){word}([^[:alnum:]]|$)").into_bytes(),
|
||||
));
|
||||
}
|
||||
query.push(')');
|
||||
}
|
||||
let _ = write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column());
|
||||
values.push(value);
|
||||
} else if let SearchValue::KeyValues(kv) = value {
|
||||
let (key, value) = kv.iter().next().unwrap();
|
||||
|
||||
|
||||
@@ -2,17 +2,12 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
backend::{
|
||||
MAX_TOKEN_LENGTH,
|
||||
postgres::{
|
||||
DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
|
||||
into_pool_error, is_timeout_error,
|
||||
},
|
||||
backend::postgres::{
|
||||
DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error,
|
||||
into_pool_error, is_timeout_error,
|
||||
},
|
||||
search::{
|
||||
IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchOperator,
|
||||
@@ -20,7 +15,7 @@ use crate::{
|
||||
},
|
||||
write::SearchIndex,
|
||||
};
|
||||
use nlp::{language::Language, tokenizers::space::SpaceTokenizer};
|
||||
use nlp::language::Language;
|
||||
use std::fmt::Write;
|
||||
use tokio_postgres::{
|
||||
IsolationLevel,
|
||||
@@ -48,19 +43,6 @@ impl PostgresStore {
|
||||
let primary_keys = index.primary_keys();
|
||||
let all_fields = index.all_fields();
|
||||
let fields = document.fields;
|
||||
// inbuxa: keyword text (addresses, contact fields, ...) is split into
|
||||
// words before it reaches the text parser, see keyword_terms().
|
||||
let keywords = primary_keys
|
||||
.iter()
|
||||
.chain(all_fields)
|
||||
.map(|field| match fields.get(field) {
|
||||
Some(SearchValue::Text {
|
||||
value,
|
||||
language: Language::None,
|
||||
}) if field.is_text() => Some(keyword_terms(value)),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let mut values = Vec::with_capacity(fields.len() + 2);
|
||||
let mut query = format!("INSERT INTO {} (", index.psql_table());
|
||||
|
||||
@@ -92,20 +74,7 @@ impl PostgresStore {
|
||||
(0, PG_UNSTEMMED_LANG)
|
||||
};
|
||||
|
||||
if let Some(keywords) = &keywords[i] {
|
||||
let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})");
|
||||
values.push(keywords as &(dyn ToSql + Sync));
|
||||
if field.sort_column().is_some() {
|
||||
let value_ref = format!("${}", values.len() + 1);
|
||||
if text_len > 255 {
|
||||
let _ = write!(&mut query, ",left({value_ref},255)");
|
||||
} else {
|
||||
let _ = write!(&mut query, ",{value_ref}");
|
||||
}
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
}
|
||||
continue;
|
||||
} else if field.is_text() {
|
||||
if field.is_text() {
|
||||
let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})");
|
||||
} else if text_len > 512 {
|
||||
query.push_str("left(");
|
||||
@@ -165,7 +134,6 @@ impl PostgresStore {
|
||||
) -> trc::Result<Vec<R>> {
|
||||
let mut query = format!("SELECT {} FROM {}", R::field().column(), index.psql_table());
|
||||
let params = self.build_filter(&mut query, filters);
|
||||
let params = params.iter().map(SqlParam::as_sql).collect::<Vec<_>>();
|
||||
if !sort.is_empty() {
|
||||
build_sort(&mut query, sort);
|
||||
}
|
||||
@@ -187,7 +155,6 @@ impl PostgresStore {
|
||||
let table = filter.index.psql_table();
|
||||
let mut where_clause = String::new();
|
||||
let params = self.build_filter(&mut where_clause, &filter.filters);
|
||||
let params = params.iter().map(SqlParam::as_sql).collect::<Vec<_>>();
|
||||
let conn = self.conn_pool.get().await.map_err(into_pool_error)?;
|
||||
let s = conn
|
||||
.prepare_cached(&format!("DELETE FROM {table}{where_clause}"))
|
||||
@@ -229,7 +196,7 @@ impl PostgresStore {
|
||||
&self,
|
||||
query: &mut String,
|
||||
filters: &'x [SearchFilter],
|
||||
) -> Vec<SqlParam<'x>> {
|
||||
) -> Vec<&'x (dyn ToSql + Sync)> {
|
||||
if filters.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
@@ -270,10 +237,6 @@ impl PostgresStore {
|
||||
|
||||
if matches!(language, Language::None) {
|
||||
let _ = write!(query, "@@ {method}('{config}', ${value_pos})");
|
||||
if let SearchValue::Text { value, .. } = value {
|
||||
values.push(SqlParam::Owned(keyword_terms(value)));
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
let _ = write!(query, "@@ ({method}('{config}', ${value_pos})");
|
||||
for fallback in [PG_FALLBACK_LANG, PG_UNSTEMMED_LANG] {
|
||||
@@ -284,18 +247,18 @@ impl PostgresStore {
|
||||
}
|
||||
query.push(')');
|
||||
}
|
||||
values.push(SqlParam::Ref(value));
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
} else if let SearchValue::KeyValues(kv) = value {
|
||||
query.push_str(field.column());
|
||||
query.push(' ');
|
||||
|
||||
let (key, value) = kv.iter().next().unwrap();
|
||||
values.push(SqlParam::Ref(key));
|
||||
values.push(key as &(dyn ToSql + Sync));
|
||||
|
||||
if !value.is_empty() {
|
||||
let _ = write!(query, "->> ${value_pos} ");
|
||||
op.write_pqsql(query, values.len() + 1);
|
||||
values.push(SqlParam::Ref(value));
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
} else {
|
||||
let _ = write!(query, " ? ${value_pos}");
|
||||
}
|
||||
@@ -304,7 +267,7 @@ impl PostgresStore {
|
||||
query.push(' ');
|
||||
|
||||
op.write_pqsql(query, value_pos);
|
||||
values.push(SqlParam::Ref(value));
|
||||
values.push(value as &(dyn ToSql + Sync));
|
||||
}
|
||||
}
|
||||
SearchFilter::And | SearchFilter::Or => {
|
||||
@@ -358,38 +321,6 @@ impl PostgresStore {
|
||||
}
|
||||
}
|
||||
|
||||
// inbuxa: PostgreSQL's text parser keeps "[email protected]" (and host names,
|
||||
// URLs, file paths, ...) as a single token, so a search for "user" or
|
||||
// "example.com" never matched an address. Keyword text is split into words the
|
||||
// same way the built-in index splits it (SpaceTokenizer: lowercase runs of
|
||||
// alphanumerics) on both the indexing and the query side, so a full address,
|
||||
// its local part, its domain and the display-name words all match, as they do
|
||||
// on the other backends.
|
||||
pub(crate) fn keyword_terms(value: &str) -> String {
|
||||
let mut terms = String::with_capacity(value.len());
|
||||
for token in SpaceTokenizer::new(value, MAX_TOKEN_LENGTH) {
|
||||
if !terms.is_empty() {
|
||||
terms.push(' ');
|
||||
}
|
||||
terms.push_str(&token);
|
||||
}
|
||||
terms
|
||||
}
|
||||
|
||||
pub(super) enum SqlParam<'x> {
|
||||
Ref(&'x (dyn ToSql + Sync)),
|
||||
Owned(String),
|
||||
}
|
||||
|
||||
impl SqlParam<'_> {
|
||||
fn as_sql(&self) -> &(dyn ToSql + Sync) {
|
||||
match self {
|
||||
SqlParam::Ref(value) => *value,
|
||||
SqlParam::Owned(value) => value,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn build_sort(query: &mut String, sort: &[SearchComparator]) {
|
||||
query.push_str(" ORDER BY ");
|
||||
for (i, comparator) in sort.iter().enumerate() {
|
||||
|
||||
@@ -21,7 +21,6 @@ pub mod replica_cluster; // inbuxa: read replicas across nodes
|
||||
pub mod scaleout; // inbuxa: scale-out storage
|
||||
#[cfg(any(feature = "postgres", feature = "mysql"))]
|
||||
pub mod sql_timeout;
|
||||
pub mod task_locks; // inbuxa: task locks across nodes
|
||||
|
||||
use crate::utils::server::TestServerBuilder;
|
||||
use std::io::Read;
|
||||
|
||||
@@ -128,11 +128,6 @@ pub async fn test(test: &TestServer) {
|
||||
println!("Running trace document tests...");
|
||||
test_trace_documents(store.clone()).await;
|
||||
|
||||
// inbuxa: address fields match by full address, local part, domain and
|
||||
// display name on every backend
|
||||
println!("Running address search tests...");
|
||||
test_address_search(store.clone()).await;
|
||||
|
||||
// Large document insert test
|
||||
println!("Running large document insert tests...");
|
||||
let mut large_text = String::with_capacity(20 * 1024 * 1024);
|
||||
@@ -977,155 +972,3 @@ async fn test_trace_documents(store: SearchStore) {
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
// inbuxa: the message indexer passes each display name and each address of
|
||||
// From/To/Cc/Bcc as keyword text (Language::None). The built-in index splits
|
||||
// that text into words, so an address is found by its full form, its local
|
||||
// part, its domain or a display-name word; PostgreSQL kept the whole address
|
||||
// as one token and MySQL dropped stopwords such as "com" and words under three
|
||||
// characters. The expected results below are the built-in (RocksDB/SQLite)
|
||||
// results and must be the same on every backend.
|
||||
async fn test_address_search(store: SearchStore) {
|
||||
const ACCOUNT_ID: u32 = 7;
|
||||
let messages: [[&[(&str, &str)]; 4]; 5] = [
|
||||
// From, To, Cc, Bcc
|
||||
[
|
||||
&[("Amazon.com", "[email protected]")],
|
||||
&[("Jane Doe", "[email protected]")],
|
||||
&[],
|
||||
&[],
|
||||
],
|
||||
[
|
||||
&[("", "[email protected]")],
|
||||
&[("", "[email protected]")],
|
||||
&[("Jane Doe", "[email protected]")],
|
||||
&[],
|
||||
],
|
||||
[
|
||||
&[("GitHub", "[email protected]")],
|
||||
&[("Jo Li", "[email protected]")],
|
||||
&[],
|
||||
&[("Audit", "[email protected]")],
|
||||
],
|
||||
[
|
||||
&[("Jane Doe", "[email protected]")],
|
||||
&[("Amazon Web Services", "[email protected]")],
|
||||
&[("Bob", "[email protected]")],
|
||||
&[("", "[email protected]")],
|
||||
],
|
||||
[
|
||||
&[("Newsletter", "[email protected]")],
|
||||
&[("", "[email protected]")],
|
||||
&[],
|
||||
&[],
|
||||
],
|
||||
];
|
||||
let fields = [
|
||||
EmailSearchField::From,
|
||||
EmailSearchField::To,
|
||||
EmailSearchField::Cc,
|
||||
EmailSearchField::Bcc,
|
||||
];
|
||||
|
||||
let mut documents = Vec::new();
|
||||
let mut mask = RoaringBitmap::new();
|
||||
for (document_id, message) in messages.iter().enumerate() {
|
||||
let mut document = IndexDocument::new(SearchIndex::Email)
|
||||
.with_account_id(ACCOUNT_ID)
|
||||
.with_document_id(document_id as u32);
|
||||
for (field, addresses) in fields.iter().zip(message.iter()) {
|
||||
for (name, address) in addresses.iter() {
|
||||
if !name.is_empty() {
|
||||
document.index_text(field.clone(), name, Language::None);
|
||||
}
|
||||
document.index_text(field.clone(), address, Language::None);
|
||||
}
|
||||
}
|
||||
document.index_unsigned(EmailSearchField::ReceivedAt, document_id as u64);
|
||||
documents.push(document);
|
||||
mask.insert(document_id as u32);
|
||||
}
|
||||
store.index(documents).await.unwrap();
|
||||
if let SearchStore::ElasticSearch(store) = &store {
|
||||
store.refresh_index(SearchIndex::Email).await.unwrap();
|
||||
}
|
||||
|
||||
for (field, text, expected) in [
|
||||
// full address
|
||||
(EmailSearchField::From, "[email protected]", vec![0u32]),
|
||||
(EmailSearchField::To, "[email protected]", vec![0]),
|
||||
(EmailSearchField::Cc, "[email protected]", vec![1]),
|
||||
(EmailSearchField::Bcc, "[email protected]", vec![3]),
|
||||
(EmailSearchField::To, "[email protected]", vec![1, 2]),
|
||||
// local part
|
||||
(EmailSearchField::From, "noreply", vec![0, 2]),
|
||||
(EmailSearchField::To, "jo", vec![1, 2]),
|
||||
(EmailSearchField::Cc, "bob", vec![3]),
|
||||
(EmailSearchField::Bcc, "audit", vec![2]),
|
||||
// domain
|
||||
(EmailSearchField::From, "amazon.com", vec![0, 1]),
|
||||
(EmailSearchField::From, "amazon", vec![0, 1]),
|
||||
(EmailSearchField::To, "example.org", vec![0, 4]),
|
||||
(EmailSearchField::To, "io.de", vec![1, 2]),
|
||||
(EmailSearchField::Cc, "example.net", vec![3]),
|
||||
(EmailSearchField::Bcc, "example.org", vec![2]),
|
||||
(EmailSearchField::From, "www.example.com", vec![4]),
|
||||
(EmailSearchField::From, "com", vec![0, 1, 2, 4]),
|
||||
// display name
|
||||
(EmailSearchField::From, "Jane", vec![3]),
|
||||
(EmailSearchField::From, "jane doe", vec![3]),
|
||||
(EmailSearchField::To, "Web Services", vec![3]),
|
||||
(EmailSearchField::To, "Li", vec![2]),
|
||||
(EmailSearchField::Cc, "Doe", vec![1]),
|
||||
(EmailSearchField::Bcc, "Audit", vec![2]),
|
||||
// hyphenated local part
|
||||
(EmailSearchField::From, "shipment-tracking", vec![1]),
|
||||
(EmailSearchField::From, "tracking", vec![1]),
|
||||
// no match
|
||||
(EmailSearchField::From, "amazon.org", vec![]),
|
||||
(EmailSearchField::To, "noreply", vec![]),
|
||||
(EmailSearchField::Bcc, "jane", vec![]),
|
||||
] {
|
||||
let ids = store
|
||||
.query_account(
|
||||
SearchQuery::new(SearchIndex::Email)
|
||||
.with_filters(vec![
|
||||
SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
|
||||
SearchFilter::has_keyword(field.clone(), text),
|
||||
])
|
||||
.with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
|
||||
.with_mask(mask.clone()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ids, expected, "{field:?} {text:?}");
|
||||
}
|
||||
|
||||
// TEXT-style search across all address fields
|
||||
let ids = store
|
||||
.query_account(
|
||||
SearchQuery::new(SearchIndex::Email)
|
||||
.with_filters(vec![
|
||||
SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID),
|
||||
SearchFilter::Or,
|
||||
SearchFilter::has_keyword(EmailSearchField::From, "example.org"),
|
||||
SearchFilter::has_keyword(EmailSearchField::To, "example.org"),
|
||||
SearchFilter::has_keyword(EmailSearchField::Cc, "example.org"),
|
||||
SearchFilter::has_keyword(EmailSearchField::Bcc, "example.org"),
|
||||
SearchFilter::End,
|
||||
])
|
||||
.with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt))
|
||||
.with_mask(mask.clone()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(ids, vec![0, 1, 2, 3, 4]);
|
||||
|
||||
store
|
||||
.unindex(
|
||||
SearchQuery::new(SearchIndex::Email)
|
||||
.with_filter(SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID)),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
@@ -1,184 +0,0 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! Task locks across nodes: tasks claimed by a node that then disappears
|
||||
//! run elsewhere once its locks expire, and a node that stops gracefully
|
||||
//! hands its locks back at once. The other node is played by writing its
|
||||
//! locks straight into the shared in-memory store, as a node that claimed
|
||||
//! the tasks and died leaves them.
|
||||
|
||||
use crate::utils::server::TestServerBuilder;
|
||||
use common::{KV_LOCK_TASK, Server};
|
||||
use registry::schema::{
|
||||
enums::IndexDocumentType,
|
||||
structs::{Task, TaskIndexDocument, TaskStatus},
|
||||
};
|
||||
use services::task_manager::lock::{TaskLockManager, release_task_locks};
|
||||
use std::time::{Duration, Instant};
|
||||
use store::{
|
||||
ValueKey,
|
||||
write::{BatchBuilder, TaskQueueClass, ValueClass},
|
||||
};
|
||||
use utils::snowflake::SnowflakeIdGenerator;
|
||||
|
||||
// Short enough for a test, long enough that the recheck interval (a twelfth
|
||||
// of it) is well below it
|
||||
const LOCK_EXPIRY: u64 = 12;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
pub async fn task_lock_tests() {
|
||||
let test = TestServerBuilder::new("task_lock_tests")
|
||||
.await
|
||||
.build()
|
||||
.await;
|
||||
let server = test.server.clone();
|
||||
println!(
|
||||
"Running task lock tests on {}...",
|
||||
std::env::var("STORE").unwrap_or_default()
|
||||
);
|
||||
server.inner.ipc.task_locks.set_expiry(LOCK_EXPIRY);
|
||||
|
||||
// 1. Another node claimed the tasks and died. Its locks block them until
|
||||
// they expire; then this node runs them, without waiting for anything
|
||||
// else to wake it
|
||||
let ids = new_task_ids(4);
|
||||
for id in &ids {
|
||||
assert!(foreign_lock(&server, *id, LOCK_EXPIRY).await);
|
||||
}
|
||||
schedule(&server, &ids).await;
|
||||
let started = Instant::now();
|
||||
server.notify_task_queue();
|
||||
tokio::time::sleep(Duration::from_secs(3)).await;
|
||||
assert_eq!(pending(&server, &ids).await, ids.len(), "held by the other node");
|
||||
wait_until_done(&server, &ids, Duration::from_secs(LOCK_EXPIRY + 10)).await;
|
||||
let elapsed = started.elapsed();
|
||||
assert!(
|
||||
elapsed >= Duration::from_secs(LOCK_EXPIRY - 2),
|
||||
"ran before the other node's locks expired: {elapsed:?}"
|
||||
);
|
||||
|
||||
// 2. The other node's locks outlive what this node expects: claimed just
|
||||
// after this node looked, or by a node whose clock runs ahead. This node
|
||||
// keeps checking at the recheck interval, so the tasks run soon after
|
||||
// those locks expire, not a whole lock lifetime later
|
||||
let held_for = LOCK_EXPIRY + LOCK_EXPIRY / 2;
|
||||
let ids = new_task_ids(4);
|
||||
for id in &ids {
|
||||
assert!(foreign_lock(&server, *id, held_for).await);
|
||||
}
|
||||
schedule(&server, &ids).await;
|
||||
let started = Instant::now();
|
||||
server.notify_task_queue();
|
||||
wait_until_done(&server, &ids, Duration::from_secs(held_for + 8)).await;
|
||||
let elapsed = started.elapsed();
|
||||
assert!(
|
||||
elapsed >= Duration::from_secs(held_for - 2),
|
||||
"ran before the other node's locks expired: {elapsed:?}"
|
||||
);
|
||||
|
||||
// 3. A graceful stop releases the locks this node holds: another node
|
||||
// can claim those tasks at once, and this one claims nothing more
|
||||
let ids = new_task_ids(3);
|
||||
for id in &ids {
|
||||
assert!(server.try_lock_task(*id).await, "claim {id}");
|
||||
}
|
||||
assert_eq!(server.inner.ipc.task_locks.held(), ids.len());
|
||||
for id in &ids {
|
||||
assert!(
|
||||
!foreign_lock(&server, *id, LOCK_EXPIRY).await,
|
||||
"held while this node runs"
|
||||
);
|
||||
}
|
||||
assert_eq!(release_task_locks(&server).await, ids.len());
|
||||
assert_eq!(server.inner.ipc.task_locks.held(), 0);
|
||||
for id in &ids {
|
||||
assert!(
|
||||
foreign_lock(&server, *id, LOCK_EXPIRY).await,
|
||||
"released on stop: {id}"
|
||||
);
|
||||
}
|
||||
let [id] = new_task_ids(1)[..] else {
|
||||
unreachable!()
|
||||
};
|
||||
assert!(!server.try_lock_task(id).await, "a stopping node claims nothing");
|
||||
|
||||
for id in ids {
|
||||
let _ = server
|
||||
.in_memory_store()
|
||||
.remove_lock(KV_LOCK_TASK, &id.to_be_bytes())
|
||||
.await;
|
||||
}
|
||||
if test.is_reset() {
|
||||
test.temp_dir.delete();
|
||||
}
|
||||
}
|
||||
|
||||
fn new_task_ids(count: usize) -> Vec<u64> {
|
||||
(0..count)
|
||||
.map(|_| SnowflakeIdGenerator::global_id().unwrap())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The other node's claim on a task, as its task manager takes it.
|
||||
async fn foreign_lock(server: &Server, id: u64, seconds: u64) -> bool {
|
||||
server
|
||||
.in_memory_store()
|
||||
.try_lock(KV_LOCK_TASK, &id.to_be_bytes(), seconds)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// Unindex tasks for files that don't exist: files aren't search-indexed and
|
||||
/// there is no undelete note, so running one only drops it from the queue.
|
||||
async fn schedule(server: &Server, ids: &[u64]) {
|
||||
let mut batch = BatchBuilder::new();
|
||||
for (n, id) in ids.iter().enumerate() {
|
||||
batch.schedule_task_with_id(
|
||||
*id,
|
||||
Task::UnindexDocument(TaskIndexDocument {
|
||||
account_id: 0u32.into(),
|
||||
document_id: (u32::MAX - n as u32).into(),
|
||||
document_type: IndexDocumentType::File,
|
||||
status: TaskStatus::now(),
|
||||
}),
|
||||
);
|
||||
}
|
||||
server.store().write(batch.build_all()).await.unwrap();
|
||||
}
|
||||
|
||||
async fn pending(server: &Server, ids: &[u64]) -> usize {
|
||||
let mut count = 0;
|
||||
for id in ids {
|
||||
if server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id: *id },
|
||||
)))
|
||||
.await
|
||||
.unwrap()
|
||||
.is_some()
|
||||
{
|
||||
count += 1;
|
||||
}
|
||||
}
|
||||
count
|
||||
}
|
||||
|
||||
async fn wait_until_done(server: &Server, ids: &[u64], within: Duration) {
|
||||
let started = Instant::now();
|
||||
loop {
|
||||
let left = pending(server, ids).await;
|
||||
if left == 0 {
|
||||
return;
|
||||
}
|
||||
assert!(
|
||||
started.elapsed() < within,
|
||||
"{left} task(s) still pending after {:?}",
|
||||
started.elapsed()
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user