feat(layer-cache) Populate S3 with any data found through S3->PG fallback

Any entries that had to be found through the S3->PG fallback logic in
the layer cache are typically from data that existed before any of the
dual-write persister modes were turned on, and were not processed by the
backfill migration mode in sdf.

The sheer volume of `cas` objects that fall into the pre-dual-write +
not-yet-backfilled category means that trying to be 100% completionist
about migrating the data from PG to S3 will take a very long time. By
migrating the data as-encountered, we can be a lot more targeted in how
we populate what's missing from S3.

Author: jhelwig

lib/dal/tests/integration_test/policy_report.rs

@@ -384,7 +384,7 @@ async fn grouped_by_name(ctx: &mut DalContext) -> Result<()> {
     assert_eq!(&policy_many_reports, &groups[0].name);
     assert_eq!(
         MAX_REPORTS_PER_GROUP as usize, // expected
-        groups[0].results.len()          // actual
+        groups[0].results.len()         // actual
     );
     // Verify all reports have the correct name
     for report in &groups[0].results {
@@ -405,16 +405,16 @@ async fn grouped_by_name(ctx: &mut DalContext) -> Result<()> {
     // Verify group 2: policy with one report
     assert_eq!(&policy_one_report, &groups[1].name);
     assert_eq!(
-        1,                           // expected
-        groups[1].results.len()      // actual
+        1,                       // expected
+        groups[1].results.len()  // actual
     );
     assert_eq!(&policy_one_report, &groups[1].results[0].name);
 
     // Verify group 3: policy with two reports
     assert_eq!(&policy_two_reports, &groups[2].name);
     assert_eq!(
-        2,                           // expected
-        groups[2].results.len()      // actual
+        2,                       // expected
+        groups[2].results.len()  // actual
     );
     for report in &groups[2].results {
         assert_eq!(&policy_two_reports, &report.name);

lib/si-layer-cache/src/layer_cache.rs

@@ -11,6 +11,12 @@ use serde::{
     de::DeserializeOwned,
 };
 use si_data_pg::PgPool;
+use si_events::{
+    Actor,
+    ChangeSetId,
+    Tenancy,
+    WorkspacePk,
+};
 use si_runtime::DedicatedExecutor;
 use telemetry::prelude::*;
 use telemetry_utils::monotonic;
@@ -24,6 +30,10 @@ use crate::{
     LayerDbError,
     db::serialize,
     error::LayerDbResult,
+    event::{
+        LayeredEvent,
+        LayeredEventKind,
+    },
     hybrid_cache::{
         Cache,
         CacheConfig,
@@ -272,6 +282,11 @@ where
 
                         let result = self.pg.get(&key).await?;
 
+                        // Write back to S3 if found in PG
+                        if let Some(ref bytes) = result {
+                            self.queue_s3_writeback(key.clone(), bytes.clone());
+                        }
+
                         let result_label: &'static str =
                             if result.is_some() { "hit" } else { "miss" };
                         monotonic!(
@@ -447,6 +462,11 @@ where
 
                         let result = self.pg.get(&key).await?;
 
+                        // Write back to S3 if found in PG
+                        if let Some(ref bytes) = result {
+                            self.queue_s3_writeback(key.clone(), bytes.clone());
+                        }
+
                         let result_label: &'static str =
                             if result.is_some() { "hit" } else { "miss" };
                         monotonic!(
@@ -557,6 +577,11 @@ where
                             );
 
                             if let Some(pg_results) = self.pg.get_many(&still_not_found).await? {
+                                // Queue write-backs for all PG-sourced data
+                                for (key, bytes) in &pg_results {
+                                    self.queue_s3_writeback(Arc::from(key.as_str()), bytes.clone());
+                                }
+
                                 // Merge S3 and PG results
                                 let mut combined = s3_results;
                                 combined.extend(pg_results);
@@ -576,7 +601,16 @@ where
                             to_backend = BackendType::Postgres.as_ref()
                         );
 
-                        self.pg.get_many(&not_found).await?
+                        let pg_results = self.pg.get_many(&not_found).await?;
+
+                        // Queue write-backs for all PG-sourced data
+                        if let Some(ref results) = pg_results {
+                            for (key, bytes) in results {
+                                self.queue_s3_writeback(Arc::from(key.as_str()), bytes.clone());
+                            }
+                        }
+
+                        pg_results
                     }
                 }
 
@@ -688,4 +722,51 @@ where
 
         s3_layer.put_direct(key, value).await
     }
+
+    /// Queue an async write-back to S3 for data found during PG fallback.
+    /// This is a fire-and-forget operation - failures are logged but don't affect the read.
+    fn queue_s3_writeback(&self, key: Arc<str>, value: Vec<u8>) {
+        // Only queue if we have S3 layers configured
+        let Some(s3_layers) = &self.s3_layers else {
+            return;
+        };
+
+        let Some(s3_layer) = s3_layers.get(self.name.as_str()) else {
+            return;
+        };
+
+        monotonic!(
+            layer_cache_fallback_writeback_attempted = 1,
+            cache_name = self.name.as_str(),
+            backend = BackendType::S3.as_ref()
+        );
+
+        // Create a write-back event with synthetic metadata.
+        // This is a caching operation, not a domain event.
+        // The S3Layer will automatically transform the key to the correct S3 format.
+        let event = LayeredEvent::new(
+            LayeredEventKind::Raw,
+            Arc::new(self.name.clone()),
+            key, // Original PG key - S3Layer transforms it
+            Arc::new(value),
+            Arc::new(String::new()), // Empty sort key - not used by S3Layer
+            None,                    // No web events for write-back
+            Tenancy::new(WorkspacePk::NONE, ChangeSetId::new()),
+            Actor::System,
+        );
+
+        // Queue for async S3 write - log on failure but don't block read
+        if let Err(e) = s3_layer.insert(event) {
+            warn!(
+                cache.name = self.name.as_str(),
+                error = ?e,
+                "failed to queue S3 write-back for fallback data"
+            );
+            monotonic!(
+                layer_cache_fallback_writeback_failed = 1,
+                cache_name = self.name.as_str(),
+                backend = BackendType::S3.as_ref()
+            );
+        }
+    }
 }