{"metadata": {"vulnerability": "CVE-2021-32714", "count": 32, "kinds": {"milestone": 8, "sighting": 3, "kev": 0, "other": 21}}, "data": [{"timestamp": "2021-05-12T00:00:00Z", "date_only": false, "value": "2021-05-12T00:00:00.000Z", "kind": "milestone", "label": "Reserved", "source": "cvelistv5", "record": "CVE-2021-32714", "field": "cveMetadata.dateReserved"}, {"timestamp": "2021-07-02T00:00:00Z", "date_only": false, "value": "2021-07-02T00:00:00.000Z", "kind": "milestone", "label": "Published", "source": "csaf_microsoft \u00b7 MSRC_CVE-2021-32714", "record": "MSRC_CVE-2021-32714", "field": "document.tracking.initial_release_date"}, {"timestamp": "2021-07-07T12:00:00Z", "date_only": false, "value": "2021-07-07T12:00:00Z", "kind": "milestone", "label": "Published", "source": "osv_rustsec \u00b7 RUSTSEC-2021-0079", "record": "RUSTSEC-2021-0079", "field": "published"}, {"timestamp": "2021-07-07T19:35:10Z", "date_only": false, "value": "2021-07-07T19:35:10.000Z", "kind": "milestone", "label": "Published", "source": "cvelistv5", "record": "CVE-2021-32714", "field": "cveMetadata.datePublished"}, {"timestamp": "2021-07-07T19:35:10Z", "date_only": false, "value": "2021-07-07T19:35:10.000Z", "kind": "other", "label": "Updated", "source": "cvelistv5 / GitHub_M", "record": "CVE-2021-32714", "field": "containers.cna.providerMetadata.dateUpdated"}, {"timestamp": "2021-07-07T20:15:00Z", "date_only": false, "value": "2021-07-07T20:15:00Z", "kind": "other", "label": "Nvd published at", "source": "github \u00b7 GHSA-5H46-H7HH-C6X9", "record": "GHSA-5H46-H7HH-C6X9", "field": "database_specific.nvd_published_at"}, {"timestamp": "2021-07-07T20:15:00Z", "date_only": false, "value": "2021-07-07T20:15:00Z", "kind": "milestone", "label": "Published", "source": "osv_ubuntu \u00b7 UBUNTU-CVE-2021-32714", "record": "UBUNTU-CVE-2021-32714", "field": "published"}, {"timestamp": "2021-07-07T20:15:00Z", "date_only": false, "value": "2021-07-07T20:15Z", "kind": "other", "label": "Published Date", "source": "gsd \u00b7 GSD-2021-32714", "record": "GSD-2021-32714", "field": "namespaces.nvd.nist.gov.publishedDate"}, {"timestamp": "2021-07-07T20:15:08.650Z", "date_only": false, "value": "2021-07-07T20:15:08.650", "kind": "milestone", "label": "Published", "source": "fkie_nvd \u00b7 FKIE_CVE-2021-32714", "record": "FKIE_CVE-2021-32714", "field": "published"}, {"timestamp": "2021-07-08T00:38:05Z", "date_only": false, "value": "2021-07-08T00:38:05.000000Z", "kind": "sighting", "label": "Seen", "source": "https://t.me/cibsecurity/25990", "record": "CVE-2021-32714", "field": "creation_timestamp", "uuid": "02910c6e-2bf6-4051-aea8-cb6d593fa164", "content": "\u203c CVE-2021-32714 \u203c\n\nhyper is an HTTP library for Rust. In versions prior to 0.14.10, hyper's HTTP server and client code had a flaw that could trigger an integer overflow when decoding chunk sizes that are too big. This allows possible data loss, or if combined with an upstream HTTP proxy that allows chunk sizes larger than hyper does, can result in \"request smuggling\" or \"desync attacks.\" The vulnerability is patched in version 0.14.10. Two possible workarounds exist. One may reject requests manually that contain a `Transfer-Encoding` header or ensure any upstream proxy rejects `Transfer-Encoding` chunk sizes greater than what fits in 64-bit unsigned integers.\n\n\ud83d\udcd6 Read\n\nvia \"National Vulnerability Database\".", "url": "https://t.me/cibsecurity/25990"}, {"timestamp": "2021-07-12T16:55:37Z", "date_only": false, "value": "2021-07-12T16:55:37Z", "kind": "milestone", "label": "Published", "source": "github \u00b7 GHSA-5H46-H7HH-C6X9", "record": "GHSA-5H46-H7HH-C6X9", "field": "published"}, {"timestamp": "2021-07-22T12:40:00Z", "date_only": false, "value": "2021-07-22T12:40Z", "kind": "other", "label": "Last Modified Date", "source": "gsd \u00b7 GSD-2021-32714", "record": "GSD-2021-32714", "field": "namespaces.nvd.nist.gov.lastModifiedDate"}, {"timestamp": "2021-09-07T21:47:36Z", "date_only": false, "value": "2021-09-07T21:47:36Z", "kind": "other", "label": "Updated", "source": "github \u00b7 GHSA-5H46-H7HH-C6X9", "record": "GHSA-5H46-H7HH-C6X9", "field": "modified"}, {"timestamp": "2023-12-13T01:23:08.677Z", "date_only": false, "value": "2023-12-13T01:23:08.677209Z", "kind": "other", "label": "Updated", "source": "gsd \u00b7 GSD-2021-32714", "record": "GSD-2021-32714", "field": "gsd.osvSchema.modified"}, {"timestamp": "2024-06-15T00:00:00Z", "date_only": false, "value": "2024-06-15T00:00:00Z", "kind": "other", "label": "Advisory revision", "source": "csaf_opensuse \u00b7 OPENSUSE-SU-2024:11751-1", "record": "OPENSUSE-SU-2024:11751-1", "field": "document.tracking.revision_history.0.date"}, {"timestamp": "2024-06-15T00:00:00Z", "date_only": false, "value": "2024-06-15T00:00:00Z", "kind": "other", "label": "Date", "source": "csaf_opensuse \u00b7 OPENSUSE-SU-2024:11751-1", "record": "OPENSUSE-SU-2024:11751-1", "field": "vulnerabilities.0.threats.0.date"}, {"timestamp": "2024-06-15T00:00:00Z", "date_only": false, "value": "2024-06-15T00:00:00Z", "kind": "other", "label": "Document generated", "source": "csaf_opensuse \u00b7 OPENSUSE-SU-2024:11751-1", "record": "OPENSUSE-SU-2024:11751-1", "field": "document.tracking.generator.date"}, {"timestamp": "2024-06-15T00:00:00Z", "date_only": false, "value": "2024-06-15T00:00:00Z", "kind": "milestone", "label": "Published", "source": "csaf_opensuse \u00b7 OPENSUSE-SU-2024:11751-1", "record": "OPENSUSE-SU-2024:11751-1", "field": "document.tracking.initial_release_date"}, {"timestamp": "2024-08-03T23:25:31.083Z", "date_only": false, "value": "2024-08-03T23:25:31.083Z", "kind": "other", "label": "Updated", "source": "cvelistv5", "record": "CVE-2021-32714", "field": "cveMetadata.dateUpdated"}, {"timestamp": "2024-08-03T23:25:31.083Z", "date_only": false, "value": "2024-08-03T23:25:31.083Z", "kind": "other", "label": "Updated", "source": "cvelistv5 / CVE", "record": "CVE-2021-32714", "field": "containers.adp.0.providerMetadata.dateUpdated"}, {"timestamp": "2024-09-11T00:00:00Z", "date_only": false, "value": "2024-09-11T00:00:00.000Z", "kind": "other", "label": "Advisory revision", "source": "csaf_microsoft \u00b7 MSRC_CVE-2021-32714", "record": "MSRC_CVE-2021-32714", "field": "document.tracking.revision_history.0.date"}, {"timestamp": "2024-09-11T00:00:00Z", "date_only": false, "value": "2024-09-11T00:00:00.000Z", "kind": "other", "label": "Date", "source": "csaf_microsoft \u00b7 MSRC_CVE-2021-32714", "record": "MSRC_CVE-2021-32714", "field": "vulnerabilities.0.remediations.0.date"}, {"timestamp": "2025-10-24T04:50:20Z", "date_only": false, "value": "2025-10-24T04:50:20Z", "kind": "other", "label": "Updated", "source": "osv_ubuntu \u00b7 UBUNTU-CVE-2021-32714", "record": "UBUNTU-CVE-2021-32714", "field": "modified"}, {"timestamp": "2026-02-18T14:49:54Z", "date_only": false, "value": "2026-02-18T14:49:54.000Z", "kind": "other", "label": "Advisory revision", "source": "csaf_microsoft \u00b7 MSRC_CVE-2021-32714", "record": "MSRC_CVE-2021-32714", "field": "document.tracking.revision_history.1.date"}, {"timestamp": "2026-02-18T14:49:54Z", "date_only": false, "value": "2026-02-18T14:49:54.000Z", "kind": "other", "label": "Updated", "source": "csaf_microsoft \u00b7 MSRC_CVE-2021-32714", "record": "MSRC_CVE-2021-32714", "field": "document.tracking.current_release_date"}, {"timestamp": "2026-02-18T15:12:04.820Z", "date_only": false, "value": "2026-02-18T15:12:04.820Z", "kind": "other", "label": "Document generated", "source": "csaf_microsoft \u00b7 MSRC_CVE-2021-32714", "record": "MSRC_CVE-2021-32714", "field": "document.tracking.generator.date"}, {"timestamp": "2026-04-07T11:43:13Z", "date_only": false, "value": "2026-04-07T11:43:13.000000Z", "kind": "sighting", "label": "Seen", "source": "https://gist.github.com/soul2zimate/a5296bc917881b1c43c396b5caf75393", "record": "CVE-2021-32714", "field": "creation_timestamp", "uuid": "b752bb41-5ec0-4e68-b684-6d227c7e31f8", "url": "https://gist.github.com/soul2zimate/a5296bc917881b1c43c396b5caf75393"}, {"timestamp": "2026-06-17T03:53:28.940Z", "date_only": false, "value": "2026-06-17T03:53:28.940", "kind": "other", "label": "Updated", "source": "fkie_nvd \u00b7 FKIE_CVE-2021-32714", "record": "FKIE_CVE-2021-32714", "field": "lastModified"}, {"timestamp": "2026-09-20T20:24:54Z", "date_only": false, "value": "2026-09-20T20:24:54Z", "kind": "other", "label": "Advisory revision", "source": "csaf_opensuse \u00b7 OPENSUSE-SU-2024:11751-1", "record": "OPENSUSE-SU-2024:11751-1", "field": "document.tracking.revision_history.1.date"}, {"timestamp": "2026-09-20T20:24:54Z", "date_only": false, "value": "2026-09-20T20:24:54Z", "kind": "other", "label": "Updated", "source": "csaf_opensuse \u00b7 OPENSUSE-SU-2024:11751-1", "record": "OPENSUSE-SU-2024:11751-1", "field": "document.tracking.current_release_date"}, {"timestamp": "2026-10-12T17:47:59Z", "date_only": false, "value": "2026-10-12T17:47:59Z", "kind": "other", "label": "Updated", "source": "osv_rustsec \u00b7 RUSTSEC-2021-0079", "record": "RUSTSEC-2021-0079", "field": "modified"}, {"timestamp": "2026-10-13T09:58:56Z", "date_only": false, "value": "2026-10-13T09:58:56.000000Z", "kind": "sighting", "label": "Seen", "source": "https://gist.github.com/dejanb/716fa74b0a444522260626adcb70e189", "record": "CVE-2021-32714", "field": "creation_timestamp", "uuid": "17657d68-7707-4c10-8f93-a6b718dba505", "content": "diff --git a/common/src/db/chunk.rs b/common/src/db/chunk.rs\nindex 28f2cbe1..2b99883e 100644\n--- a/common/src/db/chunk.rs\n+++ b/common/src/db/chunk.rs\n@@ -54,6 +54,124 @@ where\n #[cfg(test)]\n mod test {\n     use super::*;\n+    use std::{cell::Cell, rc::Rc};\n+\n+    const BIND_BUDGET: usize = (u16::MAX - 128) as usize;\n+\n+    fn assert_send(value: T) -> T {\n+        value\n+    }\n+\n+    #[test]\n+    fn chunk_empty() {\n+        assert_eq!(chunked_with(1, Vec::::new()).next(), None);\n+    }\n+\n+    #[test]\n+    fn chunk_exact_partial_multiple_order_and_duplicates() {\n+        // Three items per statement; the duplicate must not be discarded.\n+        let values = BIND_BUDGET / 3;\n+        for (input, expected) in [\n+            (vec![1, 2, 2], vec![vec![1, 2, 2]]),\n+            (vec![1, 2], vec![vec![1, 2]]),\n+            (\n+                vec![1, 2, 2, 3, 4, 5, 6],\n+                vec![vec![1, 2, 2], vec![3, 4, 5], vec![6]],\n+            ),\n+        ] {\n+            assert_eq!(chunked_with(values, input).collect::>(), expected);\n+        }\n+    }\n+\n+    #[test]\n+    fn chunk_bind_sizing_and_minimum_one() {\n+        for values in [1, 3, 17, BIND_BUDGET, BIND_BUDGET + 1] {\n+            let size = (BIND_BUDGET / values).max(1);\n+            let batches = chunked_with(values, 0..size + 1).collect::>();\n+            assert_eq!(batches[0].len(), size);\n+            assert_eq!(batches[1], vec![size]);\n+        }\n+    }\n+\n+    #[test]\n+    fn chunk_bounded_lazy_consumption() {\n+        // A non-Send input is valid: no new global bound is imposed by the helper.\n+        let consumed = Rc::new(Cell::new(0));\n+        {\n+            let input = (0..10).inspect(|_| consumed.set(consumed.get() + 1));\n+            let mut batches = chunked_with(BIND_BUDGET / 3, input);\n+            assert_eq!(consumed.get(), 0);\n+            assert_eq!(batches.next(), Some(vec![0, 1, 2]));\n+            assert_eq!(consumed.get(), 3);\n+            assert_eq!(batches.next(), Some(vec![3, 4, 5]));\n+            assert_eq!(consumed.get(), 6);\n+        }\n+        // Leaving the iterator's scope does not consume any remaining items.\n+        assert_eq!(consumed.get(), 6);\n+    }\n+\n+    mod entity {\n+        use sea_orm::entity::prelude::*;\n+\n+        #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]\n+        #[sea_orm(table_name = \"batch_test\")]\n+        pub struct Model {\n+            #[sea_orm(primary_key)]\n+            pub id: i32,\n+            pub value: String,\n+        }\n+\n+        #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]\n+        pub enum Relation {}\n+\n+        impl ActiveModelBehavior for ActiveModel {}\n+    }\n+\n+    #[test]\n+    fn chunk_entity_column_budget() {\n+        let columns = entity::Column::iter().count();\n+        let size = BIND_BUDGET / columns;\n+        let models = || {\n+            (0..size + 1).map(|_| entity::ActiveModel {\n+                ..Default::default()\n+            })\n+        };\n+        // Exercise both the free helper and the production extension-trait entry point.\n+        for batches in [\n+            chunked(models()).collect::>(),\n+            models().chunked().collect::>(),\n+        ] {\n+            assert_eq!(batches[0].len(), size);\n+            assert_eq!(batches[1].len(), 1);\n+        }\n+    }\n+\n+    #[tokio::test]\n+    async fn chunk_borrowed_non_clone_items_across_await() {\n+        struct Item(String);\n+        let items = [Item(\"first\".into()), Item(\"second\".into())];\n+        let batches = chunked_with(BIND_BUDGET, &items);\n+        assert_send(async {\n+            let mut names = Vec::new();\n+            for batch in batches {\n+                tokio::task::yield_now().await;\n+                names.extend(batch.into_iter().map(|item| item.0.as_str()));\n+            }\n+            assert_eq!(names, [\"first\", \"second\"]);\n+        })\n+        .await;\n+    }\n+\n+    #[test]\n+    fn chunk_iterator_and_awaiting_future_are_send() {\n+        let batches = assert_send(chunked_with(BIND_BUDGET / 3, vec![1, 2, 3, 4]));\n+        let _future = assert_send(async move {\n+            for batch in batches {\n+                tokio::task::yield_now().await;\n+                assert!(!batch.is_empty());\n+            }\n+        });\n+    }\n \n     #[test]\n     fn chunks() {\ndiff --git a/modules/importer/src/runner/context.rs b/modules/importer/src/runner/context.rs\nindex ddfde1c0..8d63c52e 100644\n--- a/modules/importer/src/runner/context.rs\n+++ b/modules/importer/src/runner/context.rs\n@@ -54,3 +54,35 @@ impl RunContext for () {\n     }\n     fn progress(&self, _message: String) -> impl Progress + 'static {}\n }\n+\n+#[cfg(test)]\n+mod test {\n+    use super::RunContext;\n+    use crate::runner::test::{TestRunContext, assert_send, assert_send_sync};\n+    use std::sync::atomic::{AtomicUsize, Ordering};\n+\n+    #[test_log::test(tokio::test)]\n+    async fn cancellation_factory_is_only_called_when_canceled() {\n+        assert_send_sync::();\n+        let context = TestRunContext::default();\n+        let calls = AtomicUsize::new(0);\n+        let error = || {\n+            calls.fetch_add(1, Ordering::SeqCst);\n+            \"canceled\"\n+        };\n+\n+        assert!(!assert_send(context.is_canceled()).await);\n+        assert_eq!(assert_send(context.check_canceled(error)).await, Ok(()));\n+        assert_eq!(calls.load(Ordering::SeqCst), 0);\n+\n+        context.canceled.store(true, Ordering::SeqCst);\n+        assert!(assert_send(context.is_canceled()).await);\n+        assert_eq!(\n+            assert_send(context.check_canceled(error)).await,\n+            Err(\"canceled\")\n+        );\n+        assert_eq!(calls.load(Ordering::SeqCst), 1);\n+        assert_eq!(assert_send(().check_canceled(error)).await, Ok(()));\n+        assert_eq!(calls.load(Ordering::SeqCst), 1);\n+    }\n+}\ndiff --git a/modules/importer/src/runner/csaf/mod.rs b/modules/importer/src/runner/csaf/mod.rs\nindex 3b256461..f1e6c678 100644\n--- a/modules/importer/src/runner/csaf/mod.rs\n+++ b/modules/importer/src/runner/csaf/mod.rs\n@@ -1,6 +1,9 @@\n mod report;\n pub mod storage;\n \n+#[cfg(test)]\n+mod test;\n+\n use crate::{\n     model::CsafImporter,\n     runner::{\ndiff --git a/modules/importer/src/runner/mod.rs b/modules/importer/src/runner/mod.rs\nindex 0bbdaec3..5d63d2c9 100644\n--- a/modules/importer/src/runner/mod.rs\n+++ b/modules/importer/src/runner/mod.rs\n@@ -15,6 +15,9 @@ pub mod quay;\n pub mod report;\n pub mod sbom;\n \n+#[cfg(test)]\n+pub(crate) mod test;\n+\n use crate::{\n     model::{ImporterConfiguration, auth::CredentialConfig},\n     runner::{context::RunContext, report::ScannerError},\ndiff --git a/modules/importer/src/runner/progress.rs b/modules/importer/src/runner/progress.rs\nindex 039148f0..9b801f3f 100644\n--- a/modules/importer/src/runner/progress.rs\n+++ b/modules/importer/src/runner/progress.rs\n@@ -144,8 +144,9 @@ impl TracingProgressInstance {\n #[cfg(test)]\n mod test {\n     use super::*;\n+    use crate::runner::test::{assert_send, assert_send_sync};\n     use parking_lot::Mutex;\n-    use std::sync::Arc;\n+    use std::{cell::Cell, sync::Arc};\n \n     #[derive(Clone, Default)]\n     struct MockProgress {\n@@ -189,6 +190,49 @@ mod test {\n         async fn finish(self) {}\n     }\n \n+    #[test_log::test(tokio::test)]\n+    async fn owned_progress_instance_needs_send_not_sync() {\n+        struct LocalCounter(Cell);\n+\n+        impl ProgressInstance for LocalCounter {\n+            async fn increment(&mut self, work: usize) {\n+                self.0.set(self.0.get() + work);\n+            }\n+\n+            async fn finish(self) {}\n+        }\n+\n+        // Cell is Send but not Sync. Mutable ownership is sufficient for the future.\n+        let mut progress = LocalCounter(Cell::new(0));\n+        assert_send(progress.tick()).await;\n+        assert_eq!(progress.0.get(), 1);\n+        assert_send(progress.finish()).await;\n+    }\n+\n+    #[test_log::test(tokio::test)]\n+    async fn tracing_progress_futures_tick_clamp_and_finish() {\n+        assert_send_sync::();\n+        let reporter = TracingProgress {\n+            name: \"test\".into(),\n+            period: Duration::ZERO,\n+        };\n+        let mut progress = reporter.start(2);\n+        assert_eq!(progress.current, 0);\n+        assert_eq!(progress.eta(), None);\n+\n+        assert_send(progress.tick()).await;\n+        assert_eq!(progress.current, 1);\n+        assert_send(progress.increment(10)).await;\n+        assert_eq!(progress.current, 2);\n+        assert_eq!(progress.eta(), Some(Duration::ZERO));\n+        assert_send(progress.finish()).await;\n+\n+        let mut empty = reporter.start(0);\n+        assert_send(empty.tick()).await;\n+        assert_eq!(empty.current, 0);\n+        assert_send(empty.finish()).await;\n+    }\n+\n     #[test_log::test(tokio::test)]\n     async fn simple() {\n         let reporter = MockProgress::default();\ndiff --git a/modules/importer/src/runner/sbom/mod.rs b/modules/importer/src/runner/sbom/mod.rs\nindex d6462e19..c915c139 100644\n--- a/modules/importer/src/runner/sbom/mod.rs\n+++ b/modules/importer/src/runner/sbom/mod.rs\n@@ -1,6 +1,9 @@\n mod report;\n pub mod storage;\n \n+#[cfg(test)]\n+mod test;\n+\n use crate::{\n     model::SbomImporter,\n     runner::{\ndiff --git a/modules/importer/src/server/context.rs b/modules/importer/src/server/context.rs\nindex f70001ab..b6598e3d 100644\n--- a/modules/importer/src/server/context.rs\n+++ b/modules/importer/src/server/context.rs\n@@ -144,3 +144,94 @@ where\n         // we don't support that\n     }\n }\n+\n+#[cfg(test)]\n+mod test {\n+    use super::*;\n+    use crate::runner::test::{assert_send, assert_send_sync, configuration};\n+    use test_context::test_context;\n+    use test_log::test;\n+    use trustify_common::db::{ReadWrite, pagination_cache::PaginationCache};\n+    use trustify_test_context::TrustifyContext;\n+    use uuid::Uuid;\n+\n+    #[test_context(TrustifyContext)]\n+    #[test(tokio::test)]\n+    async fn cancellation_checks_obey_cadence_and_latch(\n+        ctx: &TrustifyContext,\n+    ) -> anyhow::Result<()> {\n+        assert_send_sync::();\n+        let service =\n+            ImporterService::new(ReadWrite::new(ctx.db.clone()), PaginationCache::for_test());\n+        let name = format!(\"cancel-test-{}\", Uuid::new_v4());\n+        service.create(name.clone(), configuration()).await?;\n+        let token = CancellationToken::new();\n+        let context = ServiceRunContext::new(service.clone(), name.clone(), token.clone());\n+        assert!(!assert_send(context.is_canceled()).await);\n+        assert_eq!(\n+            assert_send(context.check_canceled(|| \"canceled\")).await,\n+            Ok(())\n+        );\n+        token.cancel();\n+        assert_eq!(\n+            assert_send(context.check_canceled(|| \"canceled\")).await,\n+            Err(\"canceled\")\n+        );\n+\n+        let period = Duration::from_secs(20);\n+        let mut check = CheckCancellation::new(\n+            service.clone(),\n+            name.clone(),\n+            period,\n+            CancellationToken::new(),\n+        );\n+        check.last_check = Instant::now() - period - Duration::from_secs(1);\n+        assert!(!assert_send(check.check()).await);\n+\n+        let mut disabled = configuration();\n+        disabled.disabled = true;\n+        service.update_configuration(&name, None, disabled).await?;\n+        check.last_check = Instant::now();\n+        let checked_at = check.last_check;\n+        assert!(\n+            !assert_send(check.check()).await,\n+            \"the cached active state lasts until the deadline\"\n+        );\n+        assert_eq!(check.last_check, checked_at);\n+        check.last_check = Instant::now() - period - Duration::from_secs(1);\n+        assert!(assert_send(check.check()).await);\n+        assert!(check.token.is_cancelled());\n+\n+        service\n+            .update_configuration(&name, None, configuration())\n+            .await?;\n+        check.last_check = Instant::now() - period - Duration::from_secs(1);\n+        let canceled_at = check.last_check;\n+        assert!(\n+            assert_send(check.check()).await,\n+            \"cancellation is latched even after re-enabling\"\n+        );\n+        assert_eq!(\n+            check.last_check, canceled_at,\n+            \"canceled checks do not query again\"\n+        );\n+\n+        let mut deleted = CheckCancellation::new(\n+            service.clone(),\n+            name.clone(),\n+            period,\n+            CancellationToken::new(),\n+        );\n+        service.delete(&name, None).await?;\n+        deleted.last_check = Instant::now();\n+        let checked_at = deleted.last_check;\n+        assert!(!assert_send(deleted.check()).await);\n+        assert_eq!(deleted.last_check, checked_at);\n+        deleted.last_check = Instant::now() - period - Duration::from_secs(1);\n+        assert!(\n+            assert_send(deleted.check()).await,\n+            \"a deleted importer cancels on the next check\"\n+        );\n+        Ok(())\n+    }\n+}\ndiff --git a/modules/importer/src/server/progress.rs b/modules/importer/src/server/progress.rs\nindex 8762fb24..4aa01d1d 100644\n--- a/modules/importer/src/server/progress.rs\n+++ b/modules/importer/src/server/progress.rs\n@@ -99,3 +99,105 @@ impl ProgressInstance for ServiceProgressInstance {\n         self.tracing.finish().await;\n     }\n }\n+\n+#[cfg(test)]\n+mod test {\n+    use super::*;\n+    use crate::runner::test::{NonSendDisplay, assert_send, assert_send_sync, configuration};\n+    use sea_orm::EntityTrait;\n+    use std::{cell::Cell, rc::Rc};\n+    use test_context::test_context;\n+    use test_log::test;\n+    use trustify_common::db::{ReadWrite, pagination_cache::PaginationCache};\n+    use trustify_entity::importer;\n+    use trustify_test_context::TrustifyContext;\n+    use uuid::Uuid;\n+\n+    fn service(ctx: &TrustifyContext) -> ImporterService {\n+        ImporterService::new(ReadWrite::new(ctx.db.clone()), PaginationCache::for_test())\n+    }\n+\n+    async fn record(ctx: &TrustifyContext, name: &str) -> anyhow::Result {\n+        Ok(importer::Entity::find_by_id(name)\n+            .one(&ctx.db)\n+            .await?\n+            .expect(\"test importer exists\"))\n+    }\n+\n+    #[test_context(TrustifyContext)]\n+    #[test(tokio::test)]\n+    async fn message_accepts_non_send_display_and_persists_when_polled(\n+        ctx: &TrustifyContext,\n+    ) -> anyhow::Result<()> {\n+        assert_send_sync::();\n+        let service = service(ctx);\n+        let name = format!(\"progress-test-{}\", Uuid::new_v4());\n+        service.create(name.clone(), configuration()).await?;\n+        let reporter = ServiceProgress::new(name.clone(), service.clone());\n+        let formats = Rc::new(Cell::new(0));\n+        // The message future need not be Send; formatting timing is not an extra contract.\n+        let future = reporter.message(NonSendDisplay(formats.clone()));\n+        assert_eq!(record(ctx, &name).await?.progress_message, None);\n+        future.await;\n+        assert_eq!(formats.get(), 1);\n+        assert_eq!(\n+            record(ctx, &name).await?.progress_message.as_deref(),\n+            Some(\"formatted progress\")\n+        );\n+        assert!(service.delete(&name, None).await?);\n+        Ok(())\n+    }\n+\n+    #[test_context(TrustifyContext)]\n+    #[test(tokio::test)]\n+    async fn tick_flush_cadence_clamps_and_finish_flushes(\n+        ctx: &TrustifyContext,\n+    ) -> anyhow::Result<()> {\n+        let service = service(ctx);\n+        let name = format!(\"progress-test-{}\", Uuid::new_v4());\n+        service.create(name.clone(), configuration()).await?;\n+        let reporter = ServiceProgress::new(name.clone(), service.clone());\n+        let mut progress = reporter.start(3);\n+        assert_eq!(record(ctx, &name).await?.progress_current, None);\n+\n+        assert_send(progress.tick()).await;\n+        let first = record(ctx, &name).await?;\n+        assert_eq!(\n+            (first.progress_current, first.progress_total),\n+            (Some(1), Some(3))\n+        );\n+\n+        // No sleeps: exercise both sides of the private flush deadline directly.\n+        progress.last_flush = Instant::now();\n+        assert_send(progress.tick()).await;\n+        assert_eq!(progress.current, 2);\n+        assert_eq!(record(ctx, &name).await?.progress_current, Some(1));\n+\n+        progress.last_flush = Instant::now() - FLUSH_PERIOD - Duration::from_secs(1);\n+        assert_send(progress.increment(10)).await;\n+        assert_eq!(progress.current, 12);\n+        assert_eq!(record(ctx, &name).await?.progress_current, Some(3));\n+        reporter.message(\"finishing\").await;\n+        assert_eq!(\n+            record(ctx, &name).await?.progress_message.as_deref(),\n+            Some(\"finishing\")\n+        );\n+        assert_send(progress.finish()).await;\n+        let finished = record(ctx, &name).await?;\n+        assert_eq!(\n+            (finished.progress_current, finished.progress_total),\n+            (Some(3), Some(3))\n+        );\n+        assert_eq!(finished.progress_message, None);\n+\n+        // Finish must flush even if the final work was below the next deadline.\n+        let mut incomplete = reporter.start(3);\n+        incomplete.last_flush = Instant::now();\n+        assert_send(incomplete.tick()).await;\n+        assert_eq!(incomplete.current, 1);\n+        assert_send(incomplete.finish()).await;\n+        assert_eq!(record(ctx, &name).await?.progress_current, Some(3));\n+        assert!(service.delete(&name, None).await?);\n+        Ok(())\n+    }\n+}\ndiff --git a/modules/importer/src/runner/csaf/test.rs b/modules/importer/src/runner/csaf/test.rs\nnew file mode 100644\nindex 00000000..f93d37c8\n--- /dev/null\n+++ b/modules/importer/src/runner/csaf/test.rs\n@@ -0,0 +1,245 @@\n+use super::{report::CsafReportVisitor, storage::StorageVisitor};\n+use crate::runner::{\n+    common::storage::StorageError,\n+    report::{Phase, ReportBuilder, ReportVisitor, Severity},\n+    test::{\n+        TestRunContext, assert_message, assert_not_persisted, assert_send, assert_send_sync,\n+        persisted_document,\n+    },\n+};\n+use bytes::Bytes;\n+use csaf_walker::{\n+    discover::{DiscoveredAdvisory, DiscoveredContext, DistributionContext},\n+    model::metadata::ProviderMetadata,\n+    retrieve::{RetrievalContext, RetrievedAdvisory},\n+    source::HttpSource,\n+    validation::{ValidatedAdvisory, ValidatedVisitor, ValidationContext, ValidationError},\n+};\n+use parking_lot::Mutex;\n+use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};\n+use std::{\n+    sync::{Arc, atomic::Ordering},\n+    time::SystemTime,\n+};\n+use test_context::test_context;\n+use test_log::test;\n+use tokio::task::LocalSet;\n+use trustify_common::db::ReadWrite;\n+use trustify_entity::{advisory, labels::Labels};\n+use trustify_test_context::{TrustifyContext, document_bytes};\n+use url::Url;\n+use uuid::Uuid;\n+\n+const SOURCE: &str = \"https://example.test/csaf/\";\n+const FILE: &str = \"https://example.test/csaf/test.json\";\n+\n+fn visitor(ctx: &TrustifyContext, context: TestRunContext) -> CsafReportVisitor {\n+    let report = Arc::new(Mutex::new(ReportBuilder::new()));\n+    CsafReportVisitor {\n+        next: ReportVisitor::new(\n+            report.clone(),\n+            StorageVisitor {\n+                context,\n+                labels: Labels::from_one(\"test\", \"send-callback\"),\n+                ingestor: ctx.ingestor.clone(),\n+                db: ReadWrite::new(ctx.db.clone()),\n+                report,\n+            },\n+        ),\n+        ignore_errors: Default::default(),\n+    }\n+}\n+\n+async fn document(tag: &str) -> anyhow::Result {\n+    let data = document_bytes(\"csaf/vex-cve-2023-0044-backport.json\").await?;\n+    let mut value: serde_json::Value = serde_json::from_slice(&data)?;\n+    let tag = format!(\"{tag}-{}\", Uuid::new_v4());\n+    value[\"document\"][\"tracking\"][\"id\"] = tag.clone().into();\n+    value[\"document\"][\"title\"] = tag.into();\n+    Ok(ValidatedAdvisory {\n+        retrieved: RetrievedAdvisory {\n+            discovered: DiscoveredAdvisory {\n+                context: Arc::new(DistributionContext::Directory(Url::parse(SOURCE)?)),\n+                url: Url::parse(FILE)?,\n+                digest: None,\n+                signature: None,\n+                modified: SystemTime::UNIX_EPOCH,\n+            },\n+            data: Bytes::from(serde_json::to_vec(&value)?),\n+            signature: None,\n+            sha256: None,\n+            sha512: None,\n+            metadata: Default::default(),\n+        },\n+    })\n+}\n+\n+async fn assert_persisted(ctx: &TrustifyContext, data: &[u8]) -> anyhow::Result<()> {\n+    let id = persisted_document(ctx, data).await?;\n+    let model = advisory::Entity::find()\n+        .filter(advisory::Column::SourceDocumentId.eq(id))\n+        .one(&ctx.db)\n+        .await?\n+        .expect(\"callback must persist the advisory\");\n+    assert_eq!(\n+        model.labels,\n+        Labels::new()\n+            .add(\"source\", SOURCE)\n+            .add(\"importer\", \"callback-test\")\n+            .add(\"file\", \"test.json\")\n+            .add(\"test\", \"send-callback\")\n+            .add(\"type\", \"csaf\")\n+    );\n+    Ok(())\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn storage_and_report_callbacks_are_send_and_persist(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    LocalSet::new().run_until(async {\n+        assert_send_sync::>();\n+        assert_send_sync::>>();\n+        assert_send_sync::>();\n+        let context = TestRunContext::default();\n+        let visitor = visitor(ctx, context.clone());\n+        let metadata: ProviderMetadata = serde_json::from_value(serde_json::json!({\n+            \"canonical_url\": \"https://example.test/csaf/provider-metadata.json\",\n+            \"last_updated\": \"2024-01-01T00:00:00Z\",\n+            \"metadata_version\": \"2.0\",\n+            \"publisher\": { \"category\": \"vendor\", \"name\": \"Test\", \"namespace\": \"https://example.test\" }\n+        }))?;\n+        let discovered = DiscoveredContext { metadata: &metadata };\n+        let keys = vec![];\n+        let retrieval = RetrievalContext { discovered: &discovered, keys: &keys };\n+        let validation = ValidationContext { retrieval: &retrieval };\n+        assert_send(\n+             as ValidatedVisitor>::visit_context(\n+                &visitor.next.next, &validation,\n+            ),\n+        ).await?;\n+        assert_send(visitor.visit_context(&validation)).await?;\n+\n+        let doc = document(\"storage-success\").await?;\n+        let data = doc.data.clone();\n+        assert_send(visitor.next.next.visit_advisory(&(), Ok::<_, ValidationError>(doc))).await?;\n+        assert_persisted(ctx, &data).await?;\n+        assert_eq!(visitor.next.report.lock().clone().build().number_of_items, 0);\n+\n+        let doc = document(\"report-success\").await?;\n+        let data = doc.data.clone();\n+        assert_send(visitor.visit_advisory(&(), Ok(doc))).await?;\n+        assert_persisted(ctx, &data).await?;\n+        let report = visitor.next.report.lock().clone().build();\n+        assert_eq!(report.number_of_items, 1);\n+        assert!(report.messages.is_empty());\n+        assert_eq!(context.checks.load(Ordering::SeqCst), 2);\n+        Ok(())\n+    }).await\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn cancellation_propagates_after_storage_and_report_persist(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    LocalSet::new()\n+        .run_until(async {\n+            for through_report in [false, true] {\n+                let context = TestRunContext::default();\n+                context.canceled.store(true, Ordering::SeqCst);\n+                let visitor = visitor(ctx, context.clone());\n+                let doc = document(if through_report {\n+                    \"report-canceled\"\n+                } else {\n+                    \"storage-canceled\"\n+                })\n+                .await?;\n+                let data = doc.data.clone();\n+                let result = if through_report {\n+                    assert_send(visitor.visit_advisory(&(), Ok(doc))).await\n+                } else {\n+                    assert_send(\n+                        visitor\n+                            .next\n+                            .next\n+                            .visit_advisory(&(), Ok::<_, ValidationError>(doc)),\n+                    )\n+                    .await\n+                };\n+                assert!(matches!(result, Err(StorageError::Canceled)));\n+                assert_persisted(ctx, &data).await?;\n+                assert_eq!(context.checks.load(Ordering::SeqCst), 1);\n+                let report = visitor.next.report.lock().clone().build();\n+                assert_eq!(report.number_of_items, usize::from(through_report));\n+                assert!(report.messages.is_empty());\n+            }\n+            Ok(())\n+        })\n+        .await\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn report_counts_ignored_validation_and_upload_errors(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    LocalSet::new()\n+        .run_until(async {\n+            let context = TestRunContext::default();\n+            let visitor = visitor(ctx, context.clone());\n+            let doc = document(\"errors\").await?;\n+            let data = doc.data.clone();\n+            let digest_error = || ValidationError::::DigestMismatch {\n+                expected: \"expected\".into(),\n+                actual: \"actual\".into(),\n+                retrieved: doc.retrieved.clone(),\n+            };\n+            assert!(matches!(\n+                assert_send(visitor.next.next.visit_advisory(&(), Err(digest_error()))).await,\n+                Err(StorageError::Validation(_))\n+            ));\n+            assert_send(visitor.visit_advisory(&(), Err(digest_error()))).await?;\n+            let report = visitor.next.report.lock().clone().build();\n+            assert_eq!(report.number_of_items, 1);\n+            assert_message(\n+                &report,\n+                Phase::Validation,\n+                FILE,\n+                Severity::Error,\n+                \"digest mismatch\",\n+            );\n+\n+            let mut invalid = doc;\n+            invalid.retrieved.discovered.url =\n+                Url::parse(\"https://example.test/csaf/invalid.json\")?;\n+            invalid.retrieved.data = Bytes::from(format!(\"not a document {}\", Uuid::new_v4()));\n+            let invalid_data = invalid.data.clone();\n+            assert!(matches!(\n+                assert_send(\n+                    visitor\n+                        .next\n+                        .next\n+                        .visit_advisory(&(), Ok::<_, ValidationError>(invalid.clone()))\n+                )\n+                .await,\n+                Err(StorageError::Storage(_))\n+            ));\n+            assert_send(visitor.visit_advisory(&(), Ok(invalid))).await?;\n+            let report = visitor.next.report.lock().clone().build();\n+            assert_eq!(report.number_of_items, 2);\n+            assert_message(\n+                &report,\n+                Phase::Upload,\n+                \"https://example.test/csaf/invalid.json\",\n+                Severity::Error,\n+                \"upload failed\",\n+            );\n+            assert_eq!(context.checks.load(Ordering::SeqCst), 0);\n+            assert_not_persisted(ctx, &data).await?;\n+            assert_not_persisted(ctx, &invalid_data).await?;\n+            Ok(())\n+        })\n+        .await\n+}\ndiff --git a/modules/importer/src/runner/sbom/test.rs b/modules/importer/src/runner/sbom/test.rs\nnew file mode 100644\nindex 00000000..08ee6e4f\n--- /dev/null\n+++ b/modules/importer/src/runner/sbom/test.rs\n@@ -0,0 +1,262 @@\n+use super::{report::SbomReportVisitor, storage::StorageVisitor};\n+use crate::runner::{\n+    common::storage::StorageError,\n+    report::{Phase, ReportBuilder, ReportVisitor, Severity},\n+    test::{\n+        TestRunContext, assert_message, assert_not_persisted, assert_send, assert_send_sync,\n+        persisted_document,\n+    },\n+};\n+use bytes::Bytes;\n+use parking_lot::Mutex;\n+use sbom_walker::{\n+    discover::{DiscoveredContext, DiscoveredSbom},\n+    model::metadata::SourceMetadata,\n+    retrieve::{RetrievalContext, RetrievedSbom},\n+    source::HttpSource,\n+    validation::{ValidatedSbom, ValidatedVisitor, ValidationContext},\n+};\n+use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};\n+use std::{\n+    sync::{Arc, atomic::Ordering},\n+    time::SystemTime,\n+};\n+use test_context::test_context;\n+use test_log::test;\n+use tokio::task::LocalSet;\n+use trustify_common::db::ReadWrite;\n+use trustify_entity::{labels::Labels, sbom};\n+use trustify_test_context::{TrustifyContext, document_bytes};\n+use url::Url;\n+use uuid::Uuid;\n+use walker_common::validate::ValidationError;\n+\n+const SOURCE: &str = \"https://example.test/sbom/\";\n+const FILE: &str = \"https://example.test/sbom/test.json\";\n+\n+fn visitor(ctx: &TrustifyContext, context: TestRunContext) -> SbomReportVisitor {\n+    let report = Arc::new(Mutex::new(ReportBuilder::new()));\n+    SbomReportVisitor {\n+        next: ReportVisitor::new(\n+            report.clone(),\n+            StorageVisitor {\n+                context,\n+                source: SOURCE.into(),\n+                max_size: None,\n+                labels: Labels::from_one(\"test\", \"send-callback\"),\n+                ingestor: ctx.ingestor.clone(),\n+                db: ReadWrite::new(ctx.db.clone()),\n+                report,\n+            },\n+        ),\n+        ignore_errors: Default::default(),\n+    }\n+}\n+\n+async fn document(tag: &str) -> anyhow::Result {\n+    let data = document_bytes(\"spdx/described-by-supplier.json\").await?;\n+    let mut value: serde_json::Value = serde_json::from_slice(&data)?;\n+    let tag = format!(\"{tag}-{}\", Uuid::new_v4());\n+    value[\"documentNamespace\"] = format!(\"https://example.test/{tag}\").into();\n+    value[\"name\"] = tag.into();\n+    Ok(ValidatedSbom {\n+        retrieved: RetrievedSbom {\n+            discovered: DiscoveredSbom {\n+                url: Url::parse(FILE)?,\n+                modified: SystemTime::UNIX_EPOCH,\n+            },\n+            data: Bytes::from(serde_json::to_vec(&value)?),\n+            signature: None,\n+            sha256: None,\n+            sha512: None,\n+            metadata: Default::default(),\n+        },\n+    })\n+}\n+\n+async fn assert_persisted(ctx: &TrustifyContext, data: &[u8]) -> anyhow::Result<()> {\n+    let id = persisted_document(ctx, data).await?;\n+    let model = sbom::Entity::find()\n+        .filter(sbom::Column::SourceDocumentId.eq(id))\n+        .one(&ctx.db)\n+        .await?\n+        .expect(\"callback must persist the SBOM\");\n+    assert_eq!(\n+        model.labels,\n+        Labels::new()\n+            .add(\"source\", SOURCE)\n+            .add(\"importer\", \"callback-test\")\n+            .add(\"file\", FILE)\n+            .add(\"test\", \"send-callback\")\n+            .add(\"type\", \"spdx\")\n+    );\n+    Ok(())\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn storage_and_report_callbacks_are_send_and_persist(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    LocalSet::new()\n+        .run_until(async {\n+            assert_send_sync::>();\n+            assert_send_sync::>>();\n+            assert_send_sync::>();\n+            let context = TestRunContext::default();\n+            let visitor = visitor(ctx, context.clone());\n+            let metadata = SourceMetadata { keys: vec![] };\n+            let discovered = DiscoveredContext {\n+                metadata: &metadata,\n+            };\n+            let keys = vec![];\n+            let retrieval = RetrievalContext {\n+                discovered: &discovered,\n+                keys: &keys,\n+            };\n+            let validation = ValidationContext {\n+                retrieval: &retrieval,\n+            };\n+            assert_send(visitor.next.next.visit_context(&validation)).await?;\n+            assert_send(visitor.visit_context(&validation)).await?;\n+\n+            let doc = document(\"storage-success\").await?;\n+            let data = doc.data.clone();\n+            assert_send(visitor.next.next.visit_sbom(&(), Ok(doc))).await?;\n+            assert_persisted(ctx, &data).await?;\n+            assert_eq!(\n+                visitor.next.report.lock().clone().build().number_of_items,\n+                0\n+            );\n+\n+            let doc = document(\"report-success\").await?;\n+            let data = doc.data.clone();\n+            assert_send(visitor.visit_sbom(&(), Ok(doc))).await?;\n+            assert_persisted(ctx, &data).await?;\n+            let report = visitor.next.report.lock().clone().build();\n+            assert_eq!(report.number_of_items, 1);\n+            assert!(\n+                !report\n+                    .messages\n+                    .values()\n+                    .flat_map(|files| files.values())\n+                    .flatten()\n+                    .any(|message| message.severity == Severity::Error)\n+            );\n+            assert_eq!(context.checks.load(Ordering::SeqCst), 2);\n+            Ok(())\n+        })\n+        .await\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn cancellation_propagates_after_storage_and_report_persist(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    LocalSet::new()\n+        .run_until(async {\n+            for through_report in [false, true] {\n+                let context = TestRunContext::default();\n+                context.canceled.store(true, Ordering::SeqCst);\n+                let visitor = visitor(ctx, context.clone());\n+                let doc = document(if through_report {\n+                    \"report-canceled\"\n+                } else {\n+                    \"storage-canceled\"\n+                })\n+                .await?;\n+                let data = doc.data.clone();\n+                let result = if through_report {\n+                    assert_send(visitor.visit_sbom(&(), Ok(doc))).await\n+                } else {\n+                    assert_send(visitor.next.next.visit_sbom(&(), Ok(doc))).await\n+                };\n+                assert!(matches!(result, Err(StorageError::Canceled)));\n+                assert_persisted(ctx, &data).await?;\n+                assert_eq!(context.checks.load(Ordering::SeqCst), 1);\n+                let report = visitor.next.report.lock().clone().build();\n+                assert_eq!(report.number_of_items, usize::from(through_report));\n+                assert!(\n+                    report\n+                        .messages\n+                        .values()\n+                        .flat_map(|files| files.values())\n+                        .flatten()\n+                        .all(|message| message.severity != Severity::Error)\n+                );\n+            }\n+            Ok(())\n+        })\n+        .await\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn report_counts_ignored_validation_upload_and_size_errors(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    LocalSet::new()\n+        .run_until(async {\n+            let context = TestRunContext::default();\n+            let mut visitor = visitor(ctx, context.clone());\n+            let doc = document(\"errors\").await?;\n+            let data = doc.data.clone();\n+            let digest_error = || ValidationError::::DigestMismatch {\n+                expected: \"expected\".into(),\n+                actual: \"actual\".into(),\n+                retrieved: doc.retrieved.clone(),\n+            };\n+            assert!(matches!(\n+                assert_send(visitor.next.next.visit_sbom(&(), Err(digest_error()))).await,\n+                Err(StorageError::Validation(_))\n+            ));\n+            assert_send(visitor.visit_sbom(&(), Err(digest_error()))).await?;\n+            let report = visitor.next.report.lock().clone().build();\n+            assert_eq!(report.number_of_items, 1);\n+            assert_message(\n+                &report,\n+                Phase::Validation,\n+                FILE,\n+                Severity::Error,\n+                \"digest mismatch\",\n+            );\n+\n+            let mut invalid = doc.clone();\n+            invalid.retrieved.discovered.url =\n+                Url::parse(\"https://example.test/sbom/invalid.json\")?;\n+            invalid.retrieved.data = Bytes::from(format!(\"not a document {}\", Uuid::new_v4()));\n+            let invalid_data = invalid.data.clone();\n+            assert!(matches!(\n+                assert_send(visitor.next.next.visit_sbom(&(), Ok(invalid.clone()))).await,\n+                Err(StorageError::Storage(_))\n+            ));\n+            assert_send(visitor.visit_sbom(&(), Ok(invalid))).await?;\n+            let report = visitor.next.report.lock().clone().build();\n+            assert_eq!(report.number_of_items, 2);\n+            assert_message(\n+                &report,\n+                Phase::Upload,\n+                \"https://example.test/sbom/invalid.json\",\n+                Severity::Error,\n+                \"upload failed\",\n+            );\n+\n+            visitor.next.next.max_size = Some(1);\n+            assert_send(visitor.visit_sbom(&(), Ok(doc))).await?;\n+            let report = visitor.next.report.lock().clone().build();\n+            assert_eq!(report.number_of_items, 3);\n+            assert_message(\n+                &report,\n+                Phase::Upload,\n+                FILE,\n+                Severity::Warning,\n+                \"size restriction\",\n+            );\n+            assert_eq!(context.checks.load(Ordering::SeqCst), 0);\n+            assert_not_persisted(ctx, &data).await?;\n+            assert_not_persisted(ctx, &invalid_data).await?;\n+            Ok(())\n+        })\n+        .await\n+}\ndiff --git a/modules/importer/src/runner/test.rs b/modules/importer/src/runner/test.rs\nnew file mode 100644\nindex 00000000..e2173f59\n--- /dev/null\n+++ b/modules/importer/src/runner/test.rs\n@@ -0,0 +1,226 @@\n+use super::{\n+    common::heartbeat::Heart,\n+    context::RunContext,\n+    report::{Phase, Report, Severity},\n+};\n+use crate::{model::ImporterConfiguration, service::ImporterService};\n+use futures::TryStreamExt;\n+use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};\n+use std::{\n+    cell::Cell,\n+    fmt::{self, Display},\n+    rc::Rc,\n+    sync::{\n+        Arc,\n+        atomic::{AtomicBool, AtomicUsize, Ordering},\n+    },\n+    time::Duration,\n+};\n+use test_context::test_context;\n+use test_log::test;\n+use tokio::{\n+    task::LocalSet,\n+    time::{sleep, timeout},\n+};\n+use tokio_util::sync::CancellationToken;\n+use trustify_common::{\n+    db::{ReadWrite, pagination_cache::PaginationCache},\n+    hashing::Digests,\n+    id::Id,\n+};\n+use trustify_entity::source_document;\n+use trustify_module_storage::service::{StorageBackend, StorageKey};\n+use trustify_test_context::TrustifyContext;\n+use uuid::Uuid;\n+\n+pub(crate) fn assert_send(value: T) -> T {\n+    value\n+}\n+\n+pub(crate) fn assert_send_sync() {}\n+\n+#[derive(Clone, Debug, Default)]\n+pub(crate) struct TestRunContext {\n+    pub(crate) canceled: Arc,\n+    pub(crate) checks: Arc,\n+}\n+\n+impl RunContext for TestRunContext {\n+    fn name(&self) -> &str {\n+        \"callback-test\"\n+    }\n+\n+    async fn is_canceled(&self) -> bool {\n+        self.checks.fetch_add(1, Ordering::SeqCst);\n+        self.canceled.load(Ordering::SeqCst)\n+    }\n+}\n+\n+// Deliberately neither Send nor Sync. Count formatting without requiring a Send message future.\n+pub(crate) struct NonSendDisplay(pub(crate) Rc>);\n+\n+impl Display for NonSendDisplay {\n+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {\n+        self.0.set(self.0.get() + 1);\n+        f.write_str(\"formatted progress\")\n+    }\n+}\n+\n+pub(crate) async fn persisted_document(ctx: &TrustifyContext, data: &[u8]) -> anyhow::Result {\n+    let digest = Id::sha256(&Digests::digest(data).sha256).value();\n+    let document = source_document::Entity::find()\n+        .filter(source_document::Column::Sha256.eq(digest))\n+        .one(&ctx.db)\n+        .await?\n+        .expect(\"callback must persist its source document\");\n+    let content = ctx\n+        .storage\n+        .retrieve(StorageKey::from_sha256(&document.sha256))\n+        .await?\n+        .expect(\"callback must persist its document bytes\");\n+    let chunks = content.try_collect::>().await?;\n+    let mut stored = Vec::with_capacity(data.len());\n+    for chunk in chunks {\n+        stored.extend_from_slice(&chunk);\n+    }\n+    assert_eq!(stored, data);\n+    Ok(document.id)\n+}\n+\n+pub(crate) async fn assert_not_persisted(ctx: &TrustifyContext, data: &[u8]) -> anyhow::Result<()> {\n+    let digest = Id::sha256(&Digests::digest(data).sha256).value();\n+    assert!(\n+        source_document::Entity::find()\n+            .filter(source_document::Column::Sha256.eq(digest))\n+            .one(&ctx.db)\n+            .await?\n+            .is_none(),\n+        \"callback must not persist the attempted source document\"\n+    );\n+    Ok(())\n+}\n+\n+pub(crate) fn assert_message(\n+    report: &Report,\n+    phase: Phase,\n+    file: &str,\n+    severity: Severity,\n+    text: &str,\n+) {\n+    let messages = &report.messages[&phase][file];\n+    assert_eq!(messages.len(), 1);\n+    assert_eq!(messages[0].severity, severity);\n+    assert!(messages[0].message.contains(text), \"{:?}\", messages[0]);\n+}\n+\n+pub(crate) fn configuration() -> ImporterConfiguration {\n+    serde_json::from_value(serde_json::json!({\n+        \"sbom\": {\n+            \"source\": \"https://example.test/sbom/\",\n+            \"period\": \"30s\",\n+            \"disabled\": false\n+        }\n+    }))\n+    .expect(\"valid test importer configuration\")\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(tokio::test)]\n+async fn retained_heart_runs_non_send_work_locks_out_stale_importer_and_cancels_on_drop(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    let db = ReadWrite::new(ctx.db.clone());\n+    let service = ImporterService::new(db.clone(), PaginationCache::for_test());\n+    let name = format!(\"heartbeat-test-{}\", Uuid::new_v4());\n+    service.create(name.clone(), configuration()).await?;\n+\n+    let result = timeout(\n+        Duration::from_secs(5),\n+        LocalSet::new().run_until(async {\n+            let original = service\n+                .read(&name)\n+                .await?\n+                .expect(\"test importer exists\")\n+                .value;\n+            assert_eq!(original.heartbeat, None);\n+            let token = CancellationToken::new();\n+            let started = Rc::new(Cell::new(0));\n+            let completed = Rc::new(Cell::new(false));\n+            let work_started = started.clone();\n+            let work_completed = completed.clone();\n+            let work_token = token.clone();\n+\n+            // Retain the actual Heart while deliberately non-Send work waits for cancellation.\n+            // work_completed is used after the await, keeping its Rc in the future state.\n+            let heart = Heart::new(\n+                original.clone(),\n+                db.clone(),\n+                async move {\n+                    work_started.set(work_started.get() + 1);\n+                    work_token.cancelled().await;\n+                    work_completed.set(true);\n+                },\n+                token.clone(),\n+            );\n+            while started.get() == 0 {\n+                assert!(heart.is_beating());\n+                sleep(Duration::from_millis(1)).await;\n+            }\n+            let locked = service\n+                .read(&name)\n+                .await?\n+                .expect(\"test importer exists\")\n+                .value;\n+            assert!(locked.heartbeat.is_some());\n+            assert_ne!(locked.heartbeat, original.heartbeat);\n+            assert!(locked.is_running());\n+            assert!(heart.is_beating());\n+            assert!(!token.is_cancelled());\n+            assert!(!completed.get());\n+\n+            // The original snapshot still has no heartbeat, so its optimistic lock must miss.\n+            let duplicate_started = Rc::new(Cell::new(0));\n+            let duplicate_work_started = duplicate_started.clone();\n+            let duplicate_token = CancellationToken::new();\n+            let duplicate_work_token = duplicate_token.clone();\n+            let duplicate = Heart::new(\n+                original,\n+                db.clone(),\n+                async move {\n+                    duplicate_work_started.set(duplicate_work_started.get() + 1);\n+                    duplicate_work_token.cancelled().await;\n+                    drop(duplicate_work_started);\n+                },\n+                duplicate_token.clone(),\n+            );\n+            while duplicate.is_beating() {\n+                sleep(Duration::from_millis(1)).await;\n+            }\n+            assert_eq!(duplicate_started.get(), 0);\n+            assert_eq!(\n+                Rc::strong_count(&duplicate_started),\n+                1,\n+                \"stale work was dropped\"\n+            );\n+            assert_eq!(started.get(), 1);\n+            assert!(heart.is_beating());\n+            drop(duplicate);\n+            assert!(duplicate_token.is_cancelled());\n+            assert!(!token.is_cancelled());\n+\n+            drop(heart);\n+            assert!(token.is_cancelled(), \"dropping Heart must cancel its token\");\n+            while !completed.get() {\n+                sleep(Duration::from_millis(1)).await;\n+            }\n+            assert_eq!(started.get(), 1);\n+            Ok::<_, anyhow::Error>(())\n+        }),\n+    )\n+    .await;\n+\n+    // Also remove our own row on an ordinary error or timeout; never touch other importers.\n+    assert!(service.delete(&name, None).await?);\n+    result??;\n+    Ok(())\n+}\ndiff --git a/modules/ingestor/tests/send_sync.rs b/modules/ingestor/tests/send_sync.rs\nnew file mode 100644\nindex 00000000..42f440f6\n--- /dev/null\n+++ b/modules/ingestor/tests/send_sync.rs\n@@ -0,0 +1,230 @@\n+#![recursion_limit = \"512\"]\n+\n+use sea_orm::{\n+    ActiveValue::Set, ColumnTrait, ConnectionTrait, DbErr, EntityTrait, Iterable, PaginatorTrait,\n+    QueryFilter, QueryOrder, Select, TransactionTrait,\n+};\n+use std::sync::atomic::{AtomicUsize, Ordering};\n+use test_context::test_context;\n+use test_log::test;\n+use trustify_common::db::chunk::chunked;\n+use trustify_entity::{advisory, sbom, vulnerability};\n+use trustify_module_ingestor::{\n+    graph::vulnerability::{VulnerabilityInformation, creator::VulnerabilityCreator},\n+    service::{Cache, Format},\n+};\n+use trustify_test_context::{TrustifyContext, document_bytes};\n+use uuid::Uuid;\n+\n+fn assert_send(future: T) -> T {\n+    future\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(actix_web::test)]\n+async fn ingest_futures_are_send_with_borrowed_transactions(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    for (format, path) in [\n+        (Format::CSAF, \"csaf/cve-2023-0044.json\"),\n+        (Format::CVE, \"cve/CVE-2021-32714.json\"),\n+        (Format::OSV, \"osv/RUSTSEC-2021-0079.json\"),\n+        (Format::SPDX, \"spdx/simple.json\"),\n+        (Format::CycloneDX, \"cyclonedx/simple_1dot6.json\"),\n+    ] {\n+        let bytes = document_bytes(path).await?;\n+        let mut value: serde_json::Value = serde_json::from_slice(&bytes)?;\n+        let name = format!(\"send-regression-{}\", Uuid::new_v4());\n+        match format {\n+            Format::CSAF => value[\"document\"][\"title\"] = name.into(),\n+            Format::CVE => value[\"containers\"][\"cna\"][\"title\"] = name.into(),\n+            Format::OSV => value[\"summary\"] = name.into(),\n+            Format::SPDX => value[\"name\"] = name.into(),\n+            Format::CycloneDX => value[\"metadata\"][\"component\"][\"name\"] = name.into(),\n+            _ => unreachable!(\"only the listed formats are tested\"),\n+        }\n+        let bytes = serde_json::to_vec(&value)?;\n+        let tx = ctx.db.begin().await?;\n+\n+        // Check the real service/dispatcher future, without requiring 'static or spawning it.\n+        let result = assert_send(ctx.ingestor.ingest(\n+            &bytes,\n+            format,\n+            (\"source\", \"send-regression\"),\n+            None,\n+            Cache::Skip,\n+            &tx,\n+        ))\n+        .await?;\n+        assert!(!result.duplicate, \"{path} should be newly ingested\");\n+        let id = Uuid::parse_str(&result.id)?;\n+        assert_document_present(&tx, format, id, true).await?;\n+        assert_document_present(&ctx.db, format, id, false).await?;\n+\n+        tx.rollback().await?;\n+        assert_document_present(&ctx.db, format, id, false).await?;\n+    }\n+\n+    Ok(())\n+}\n+\n+async fn assert_document_present(\n+    connection: &impl ConnectionTrait,\n+    format: Format,\n+    id: Uuid,\n+    present: bool,\n+) -> Result<(), DbErr> {\n+    let count = if matches!(format, Format::SPDX | Format::CycloneDX) {\n+        sbom::Entity::find_by_id(id).count(connection).await?\n+    } else {\n+        advisory::Entity::find_by_id(id).count(connection).await?\n+    };\n+    assert_eq!(count, u64::from(present), \"{format} document visibility\");\n+    Ok(())\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(actix_web::test)]\n+async fn vulnerability_creator_multiple_batches_preserve_transactions(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    let prefix = format!(\"SEND-SYNC-{}-\", Uuid::new_v4());\n+    let columns = vulnerability::Column::iter().count();\n+    let batch_size = usize::from(u16::MAX - 128) / columns;\n+    // The creator leaves authoritative_advisory_id and generated id_sort_key NotSet.\n+    // Exceed even the hard limit for the remaining bound columns, so removing chunking\n+    // would fail. Currently this is 6,556 rows, split into batches of 5,450 and 1,106.\n+    let bound_columns = columns - 2;\n+    let item_count = usize::from(u16::MAX) / bound_columns + 3;\n+    assert!(item_count > batch_size);\n+    assert_eq!(regression_vulnerabilities(&prefix).count(&ctx.db).await?, 0);\n+\n+    // Roll back actual multi-batch inserts, not just updates of existing rows.\n+    let tx = ctx.db.begin().await?;\n+    assert_send(vulnerability_creator(&prefix, item_count, None).create(&tx)).await?;\n+    assert_vulnerabilities(&tx, &prefix, item_count, None).await?;\n+    assert_eq!(regression_vulnerabilities(&prefix).count(&ctx.db).await?, 0);\n+    tx.rollback().await?;\n+    assert_eq!(regression_vulnerabilities(&prefix).count(&ctx.db).await?, 0);\n+\n+    // The creator must leave committing to its caller, even after the last batch.\n+    let tx = ctx.db.begin().await?;\n+    assert_send(vulnerability_creator(&prefix, item_count, Some(\"committed\")).create(&tx)).await?;\n+    assert_vulnerabilities(&tx, &prefix, item_count, Some(\"committed\")).await?;\n+    assert_eq!(regression_vulnerabilities(&prefix).count(&ctx.db).await?, 0);\n+    tx.commit().await?;\n+    assert_vulnerabilities(&ctx.db, &prefix, item_count, Some(\"committed\")).await?;\n+\n+    // Exercise both ON CONFLICT paths across batches: empty data must not overwrite,\n+    // while supplied data must update. Rolling back must preserve the committed data.\n+    let tx = ctx.db.begin().await?;\n+    assert_send(vulnerability_creator(&prefix, item_count, None).create(&tx)).await?;\n+    assert_vulnerabilities(&tx, &prefix, item_count, Some(\"committed\")).await?;\n+    assert_send(vulnerability_creator(&prefix, item_count, Some(\"rolled back\")).create(&tx))\n+        .await?;\n+    assert_vulnerabilities(&tx, &prefix, item_count, Some(\"rolled back\")).await?;\n+    assert_vulnerabilities(&ctx.db, &prefix, item_count, Some(\"committed\")).await?;\n+    tx.rollback().await?;\n+    assert_vulnerabilities(&ctx.db, &prefix, item_count, Some(\"committed\")).await?;\n+\n+    // Remove only this invocation's committed rows, including on a shared test database.\n+    let deleted = vulnerability::Entity::delete_many()\n+        .filter(vulnerability::Column::Id.starts_with(&prefix))\n+        .exec(&ctx.db)\n+        .await?;\n+    assert_eq!(deleted.rows_affected, item_count as u64);\n+    assert_eq!(regression_vulnerabilities(&prefix).count(&ctx.db).await?, 0);\n+    Ok(())\n+}\n+\n+#[test_context(TrustifyContext)]\n+#[test(actix_web::test)]\n+async fn chunked_later_constraint_error_rolls_back_earlier_inserts(\n+    ctx: &TrustifyContext,\n+) -> anyhow::Result<()> {\n+    let prefix = format!(\"SEND-SYNC-FAIL-{}-\", Uuid::new_v4());\n+    let batch_size = usize::from(u16::MAX - 128) / vulnerability::Column::iter().count();\n+    let missing_advisory = Uuid::new_v4();\n+    let completed = AtomicUsize::new(0);\n+    let models = (0..batch_size + 1).map(|index| vulnerability::ActiveModel {\n+        id: Set(vulnerability_id(&prefix, index)),\n+        // Only the last batch violates the authoritative-advisory foreign key.\n+        authoritative_advisory_id: Set((index == batch_size).then_some(missing_advisory)),\n+        ..Default::default()\n+    });\n+    let result = assert_send(ctx.db.transaction(async |tx| {\n+        for batch in chunked(models) {\n+            vulnerability::Entity::insert_many(batch)\n+                .exec_without_returning(tx)\n+                .await?;\n+            completed.fetch_add(1, Ordering::SeqCst);\n+        }\n+        Ok::<_, DbErr>(())\n+    }))\n+    .await;\n+    assert_eq!(completed.load(Ordering::SeqCst), 1);\n+    let error = result.expect_err(\"the second batch must fail\");\n+    assert!(error.to_string().contains(\"foreign key\"), \"{error}\");\n+    assert_eq!(regression_vulnerabilities(&prefix).count(&ctx.db).await?, 0);\n+    Ok(())\n+}\n+\n+fn vulnerability_id(prefix: &str, index: usize) -> String {\n+    format!(\"{prefix}{index:06}\")\n+}\n+\n+fn vulnerability_creator(\n+    prefix: &str,\n+    item_count: usize,\n+    title: Option<&str>,\n+) -> VulnerabilityCreator {\n+    let mut creator = VulnerabilityCreator::new();\n+    // Reverse input order and duplicate every key to exercise creator ordering/deduplication.\n+    for index in (0..item_count).rev() {\n+        let id = vulnerability_id(prefix, index);\n+        creator.add(&id, ());\n+        if let Some(title) = title {\n+            creator.add(\n+                &id,\n+                VulnerabilityInformation {\n+                    title: Some(\"superseded\".to_owned()),\n+                    ..Default::default()\n+                },\n+            );\n+            creator.add(\n+                &id,\n+                VulnerabilityInformation {\n+                    title: Some(format!(\"{title} {index}\")),\n+                    ..Default::default()\n+                },\n+            );\n+        }\n+        // A later empty entry must not discard accumulated information.\n+        creator.add(id, ());\n+    }\n+    creator\n+}\n+\n+fn regression_vulnerabilities(prefix: &str) -> Select {\n+    vulnerability::Entity::find().filter(vulnerability::Column::Id.starts_with(prefix))\n+}\n+\n+async fn assert_vulnerabilities(\n+    connection: &impl ConnectionTrait,\n+    prefix: &str,\n+    item_count: usize,\n+    title: Option<&str>,\n+) -> Result<(), DbErr> {\n+    // SQL tables have no implicit order; compare an explicitly ordered result, including\n+    // every identifier/title on both sides of the batch boundary.\n+    let rows = regression_vulnerabilities(prefix)\n+        .order_by_asc(vulnerability::Column::Id)\n+        .all(connection)\n+        .await?;\n+    assert_eq!(rows.len(), item_count);\n+    for (index, row) in rows.into_iter().enumerate() {\n+        assert_eq!(row.id, vulnerability_id(prefix, index));\n+        assert_eq!(row.title, title.map(|title| format!(\"{title} {index}\")));\n+    }\n+    Ok(())\n+}\n", "url": "https://gist.github.com/dejanb/716fa74b0a444522260626adcb70e189"}]}