fix: requeue uncommitted notifications
This commit is contained in:
@@ -254,15 +254,25 @@ impl Interceptor for WorkerInterceptor {
|
||||
};
|
||||
let mut system_items: Vec<SystemItem> = Vec::with_capacity(drained.len());
|
||||
let mut items: Vec<Item> = Vec::with_capacity(drained.len());
|
||||
for entry in drained {
|
||||
let system_item =
|
||||
build_system_item_with_provenance(&entry, &prompts, Some(provenance.clone()))
|
||||
.map_err(|error| format!("failed to render notify_wrapper: {error}"))?;
|
||||
for entry in &drained {
|
||||
let system_item = match build_system_item_with_provenance(
|
||||
entry,
|
||||
&prompts,
|
||||
Some(provenance.clone()),
|
||||
) {
|
||||
Ok(system_item) => system_item,
|
||||
Err(error) => {
|
||||
self.pending_notifies.requeue_front(drained);
|
||||
return Err(format!("failed to render notify_wrapper: {error}"));
|
||||
}
|
||||
};
|
||||
items.push(system_item.to_history_item());
|
||||
system_items.push(system_item);
|
||||
}
|
||||
self.commit_system_items(&system_items)
|
||||
.map_err(|error| format!("session persistence failed: {error}"))?;
|
||||
if let Err(error) = self.commit_system_items(&system_items) {
|
||||
self.pending_notifies.requeue_front(drained);
|
||||
return Err(format!("session persistence failed: {error}"));
|
||||
}
|
||||
Ok(items)
|
||||
}
|
||||
|
||||
@@ -1129,6 +1139,46 @@ mod tests {
|
||||
assert_eq!(provenance.logical_name, "internal.notify_wrapper");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn notify_render_failure_requeues_without_context_only_fallback() {
|
||||
let prompts = test_prompts();
|
||||
let buffer = NotifyBuffer::new();
|
||||
let interceptor = WorkerInterceptor::new(
|
||||
Arc::new(HookRegistryBuilder::new().build()),
|
||||
None,
|
||||
None,
|
||||
buffer.clone(),
|
||||
Arc::new(Mutex::new(Vec::new())),
|
||||
prompts.clone(),
|
||||
None,
|
||||
);
|
||||
let current = prompts.load_full();
|
||||
let projection = current.projection();
|
||||
let mut templates = projection.templates.clone();
|
||||
templates.insert(
|
||||
"internal.notify_wrapper".to_string(),
|
||||
"{{ message | missing_notify_filter }}".to_string(),
|
||||
);
|
||||
let mut projection = crate::prompt::catalog::EffectivePromptCatalog::new(
|
||||
templates,
|
||||
3,
|
||||
projection.schema_fingerprint.clone(),
|
||||
projection.toolchain_fingerprint.clone(),
|
||||
)
|
||||
.unwrap();
|
||||
projection.source_digest = "source-3".to_string();
|
||||
prompts.store(Arc::new(
|
||||
PromptCatalog::from_projection(projection).unwrap(),
|
||||
));
|
||||
buffer.push_notify("must persist".to_string(), false);
|
||||
|
||||
let error = interceptor.pending_history_appends().await.unwrap_err();
|
||||
|
||||
assert!(error.contains("failed to render notify_wrapper"));
|
||||
let requeued = buffer.drain();
|
||||
assert_eq!(requeued.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pending_history_appends_drains_buffer_into_items() {
|
||||
let registry = Arc::new(HookRegistryBuilder::new().build());
|
||||
|
||||
@@ -89,6 +89,23 @@ impl NotifyBuffer {
|
||||
q.drain(..).collect()
|
||||
}
|
||||
|
||||
/// Restore a failed drain ahead of entries queued concurrently while the
|
||||
/// consumer was rendering. FIFO order is preserved.
|
||||
pub(crate) fn requeue_front(&self, entries: Vec<PendingNotify>) {
|
||||
let mut q = self.inner.lock().expect("notify buffer poisoned");
|
||||
for entry in entries.into_iter().rev() {
|
||||
q.push_front(entry);
|
||||
}
|
||||
while q.len() > CAPACITY {
|
||||
let dropped = q.pop_front();
|
||||
warn!(
|
||||
capacity = CAPACITY,
|
||||
dropped = ?dropped,
|
||||
"notify buffer overflow while restoring failed drain; dropped oldest"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether an undrained `Method::Notify { auto_run: true }` remains.
|
||||
pub fn has_auto_run_pending(&self) -> bool {
|
||||
self.inner
|
||||
|
||||
Reference in New Issue
Block a user