Report reschedules keep the task queue readable
Setting deliverAt on an internal DMARC or TLS report wrote the new task
queue row with the report's object type (0x21, 0x6e) instead of the task
type (7, 8), and left the task row at its old due. The task manager's scan
failed on that row with store.data-corruption ("Failed to iterate over task
queue"), and because the error ended the whole scan, every task due after
the row stopped running on every node.
- reschedule_ops writes the new queue row through schedule_task_with_id, so
it carries the task type and the task row gets the new due. It removes
the row the task is actually queued under (the task's due, which differs
from deliverAt once the task has been retried) and any row an earlier
reschedule left at deliverAt.
- x:DmarcInternalReport/set and x:TlsInternalReport/set lock the report's
task while they move it, as x:Task/set does, refuse while the report is
being sent, release the locks however the request ends, and wake the task
manager.
- The task manager logs a queue row it can't read (id, due, key, value) and
skips it instead of ending the scan. It then repairs the row from its task:
the row is rewritten with the task's type, and a row with no task behind
it is removed. A row holding a report's object type for a report task is
what the old reschedule wrote: the task is moved to that row's time, as
the reschedule intended, and its old queue row is removed. Stores that
already hold such a row recover on their own once it comes due.
- x:Task/query with a type filter skips an unreadable row instead of
failing.
Test: smtp::reporting::reschedule (RocksDB and PostgreSQL). It fails on
main: x:Task/get shows the old due, and with that check removed, neither
report nor a later task ever runs.
This commit is contained in:
@@ -30,6 +30,7 @@ use common::network::limiter::ConcurrencyLimiter;
|
||||
use common::network::{ServerInstance, TcpAcceptor};
|
||||
use common::{Inner, Server};
|
||||
use registry::schema::enums::TaskType;
|
||||
use registry::schema::prelude::ObjectType;
|
||||
use registry::schema::structs::{
|
||||
Task, TaskManager, TaskRetryStrategy, TaskStatus, TaskStatusFailed, TaskStatusRetry,
|
||||
};
|
||||
@@ -298,6 +299,7 @@ impl TaskQueueManager for Server {
|
||||
|
||||
// Retrieve tasks pending to be processed
|
||||
let mut tasks = Vec::new();
|
||||
let mut unreadable = Vec::new();
|
||||
let now = Instant::now();
|
||||
let mut next_event = None;
|
||||
ipc.revision += 1;
|
||||
@@ -311,12 +313,21 @@ impl TaskQueueManager for Server {
|
||||
let task_id = key.deserialize_be_u64(U64_LEN)?;
|
||||
|
||||
if task_due <= now_timestamp {
|
||||
let task_type_idx = value.deserialize_be_u16(0)?;
|
||||
let task_type = TaskType::from_id(task_type_idx).ok_or_else(|| {
|
||||
trc::StoreEvent::DataCorruption
|
||||
.caused_by(trc::location!())
|
||||
.ctx(trc::Key::Value, value)
|
||||
})?;
|
||||
// inbuxa: a row whose task type can't be read is
|
||||
// set aside, not allowed to end the scan: every
|
||||
// task due after it would wait behind it
|
||||
let Some((task_type_idx, task_type)) = value
|
||||
.deserialize_be_u16(0)
|
||||
.ok()
|
||||
.and_then(|idx| TaskType::from_id(idx).map(|typ| (idx, typ)))
|
||||
else {
|
||||
unreadable.push(UnreadableDueRow {
|
||||
due: task_due,
|
||||
id: task_id,
|
||||
value: value.to_vec(),
|
||||
});
|
||||
return Ok(true);
|
||||
};
|
||||
// inbuxa: running here under a lease this node
|
||||
// renews; don't hand it to a worker again
|
||||
if task_locks.is_held(task_id) {
|
||||
@@ -389,6 +400,11 @@ impl TaskQueueManager for Server {
|
||||
);
|
||||
});
|
||||
|
||||
if !unreadable.is_empty() && repair_due_rows(self, unreadable).await {
|
||||
// Look again at once for the rows that were rewritten
|
||||
self.notify_task_queue();
|
||||
}
|
||||
|
||||
if !tasks.is_empty() {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskAcquired),
|
||||
@@ -819,3 +835,114 @@ impl TaskResult {
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// inbuxa: a task queue row whose task type could not be read.
|
||||
struct UnreadableDueRow {
|
||||
due: u64,
|
||||
id: u64,
|
||||
value: Vec<u8>,
|
||||
}
|
||||
|
||||
/// inbuxa: logs each unreadable queue row and repairs it from the task it
|
||||
/// schedules. The task row says what the task is, so the queue row is
|
||||
/// rewritten with that task's type; a row with no task behind it is removed.
|
||||
///
|
||||
/// Rescheduling an internal DMARC or TLS report wrote the report's object
|
||||
/// type into the queue row instead of the task type. Such a row is the time
|
||||
/// an administrator chose, so the task is moved to it as the reschedule
|
||||
/// meant to do: the task row takes that due, and a queue row left at the
|
||||
/// task's previous due is removed. Returns whether any row was repaired.
|
||||
async fn repair_due_rows(server: &Server, rows: Vec<UnreadableDueRow>) -> bool {
|
||||
let mut repaired = false;
|
||||
for row in rows {
|
||||
let UnreadableDueRow { due, id, value } = row;
|
||||
trc::error!(
|
||||
trc::StoreEvent::DataCorruption
|
||||
.into_err()
|
||||
.id(id)
|
||||
.ctx(trc::Key::Due, trc::Value::Timestamp(due))
|
||||
.ctx(
|
||||
trc::Key::Key,
|
||||
[due.to_be_bytes(), id.to_be_bytes()].concat()
|
||||
)
|
||||
.ctx(trc::Key::Value, value.clone())
|
||||
.details("Unreadable task queue row skipped")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
|
||||
let task_key = ValueClass::TaskQueue(TaskQueueClass::Task { id });
|
||||
let due_key = ValueClass::TaskQueue(TaskQueueClass::Due { id, due });
|
||||
let task = match server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(task_key.clone()))
|
||||
.await
|
||||
{
|
||||
Ok(task) => task,
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(id)
|
||||
.details("Failed to read the task of an unreadable queue row.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let mut batch = BatchBuilder::new();
|
||||
let action = if let Some(mut task) = task {
|
||||
let task_type = task.object_type();
|
||||
batch.assert_value(task_key.clone(), AssertValue::Some);
|
||||
if rescheduled_report_type(&value) == Some(task_type) {
|
||||
let old_due = task.due_timestamp();
|
||||
if old_due != due {
|
||||
batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id,
|
||||
due: old_due,
|
||||
}));
|
||||
}
|
||||
task.set_status(TaskStatus::at(due as i64));
|
||||
}
|
||||
batch
|
||||
.set(due_key, task_type.to_id().serialize())
|
||||
.set(task_key, task.to_pickled_vec());
|
||||
"Rewrote the queue row from its task."
|
||||
} else {
|
||||
batch.clear(due_key);
|
||||
"Removed a queue row with no task."
|
||||
};
|
||||
|
||||
match server.store().write(batch.build_all()).await {
|
||||
Ok(_) => {
|
||||
repaired = true;
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskIgnored),
|
||||
Id = id,
|
||||
Due = trc::Value::Timestamp(due),
|
||||
Reason = action,
|
||||
);
|
||||
}
|
||||
Err(err) if err.matches(trc::EventType::Store(trc::StoreEvent::AssertValueFailed)) => {
|
||||
// The task went away meanwhile; the next scan looks again
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(id)
|
||||
.details("Failed to repair an unreadable queue row.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
repaired
|
||||
}
|
||||
|
||||
/// inbuxa: the task type a report reschedule meant, when a queue row holds
|
||||
/// an internal report's object type (the value that reschedule wrote).
|
||||
fn rescheduled_report_type(value: &[u8]) -> Option<TaskType> {
|
||||
let id = u16::from_be_bytes(value.get(..2)?.try_into().ok()?);
|
||||
match ObjectType::from_id(id)? {
|
||||
ObjectType::DmarcInternalReport => Some(TaskType::DmarcReport),
|
||||
ObjectType::TlsInternalReport => Some(TaskType::TlsReport),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user