Turbopack: use the same serialization method for lookup as for storing (#84754)

### What?

This makes sure to use the same method for serializing CachedTaskType
for storing and lookup.
Before that we instantiated the serialization twice and that caused some
weird behavior on linux.

Before this fix I was seeing that serialization of CachedTaskType was
not byte to byte identical between storing and lookup. This caused tasks
to not be found, which will lead to very weird effects: stale tasks,
cell not found, etc.

I was expecting that pot serialization is deterministic, but turns out
it isn't fully deterministic. It has some logic to deduplicate Symbols
(like enum variant names). But it deduplicates these `&'static str`s
based on the pointer address of the static string. Turns out that Rust
doesn't guarantee that static strings are deduplicated in all cases.

I assume that we are somehow running into that problem when having two
variants of the serialization. They ended up serializing task types
differently (one didn't have a symbol deduplicated in the serialized
form. But that's more a wild guess. It did fix the bug, but I have a
hard time understanding why this is happening.

On long term, I think we should change pot to avoid using pointer
addresses and do a real comparision instead.

Author: sokra

turbopack/crates/turbo-tasks-backend/src/kv_backing_storage.rs

@@ -357,8 +357,12 @@ impl<T: KeyValueDatabase + Send + Sync + 'static> BackingStorageSealed
 
                             let mut task_type_bytes = Vec::new();
                             for (task_type, task_id) in updates {
+                                serialize_task_type(
+                                    &task_type,
+                                    &mut task_type_bytes,
+                                    Some(task_id),
+                                )?;
                                 let task_id: u32 = *task_id;
-                                serialize_task_type(&task_type, &mut task_type_bytes, task_id)?;
 
                                 batch
                                     .put(
@@ -441,8 +445,8 @@ impl<T: KeyValueDatabase + Send + Sync + 'static> BackingStorageSealed
                     .entered();
                     let mut task_type_bytes = Vec::new();
                     for (task_type, task_id) in task_cache_updates.into_iter().flatten() {
+                        serialize_task_type(&task_type, &mut task_type_bytes, Some(task_id))?;
                         let task_id = *task_id;
-                        serialize_task_type(&task_type, &mut task_type_bytes, task_id)?;
 
                         batch
                             .put(
@@ -499,8 +503,10 @@ impl<T: KeyValueDatabase + Send + Sync + 'static> BackingStorageSealed
             tx: &D::ReadTransaction<'_>,
             task_type: &CachedTaskType,
         ) -> Result<Option<TaskId>> {
-            let task_type = POT_CONFIG.serialize(task_type)?;
-            let Some(bytes) = database.get(tx, KeySpace::ForwardTaskCache, &task_type)? else {
+            let mut task_type_bytes = Vec::new();
+            serialize_task_type(task_type, &mut task_type_bytes, None)?;
+            let Some(bytes) = database.get(tx, KeySpace::ForwardTaskCache, &task_type_bytes)?
+            else {
                 return Ok(None);
             };
             let bytes = bytes.borrow().try_into()?;
@@ -645,23 +651,30 @@ where
     Ok(())
 }
 
+#[inline(never)]
 fn serialize_task_type(
-    task_type: &Arc<CachedTaskType>,
+    task_type: &CachedTaskType,
     mut task_type_bytes: &mut Vec<u8>,
-    task_id: u32,
+    task_id: Option<TaskId>,
 ) -> Result<()> {
     task_type_bytes.clear();
     POT_CONFIG
-        .serialize_into(&**task_type, &mut task_type_bytes)
-        .with_context(|| anyhow!("Unable to serialize task {task_id} cache key {task_type:?}"))?;
+        .serialize_into(task_type, &mut task_type_bytes)
+        .with_context(|| {
+            if let Some(task_id) = task_id {
+                anyhow!("Unable to serialize task {task_id} cache key {task_type:?}")
+            } else {
+                anyhow!("Unable to serialize task cache key {task_type:?}")
+            }
+        })?;
     #[cfg(feature = "verify_serialization")]
     {
         let deserialize: Result<CachedTaskType, _> = serde_path_to_error::deserialize(
             &mut pot_de_symbol_list().deserializer_for_slice(&*task_type_bytes)?,
         );
         if let Err(err) = deserialize {
-            println!("Task type would not be deserializable {task_id}: {err:?}\n{task_type:#?}");
-            panic!("Task type would not be deserializable {task_id}: {err:?}");
+            println!("Task type would not be deserializable {task_id:?}: {err:?}\n{task_type:#?}");
+            panic!("Task type would not be deserializable {task_id:?}: {err:?}");
         }
     }
     Ok(())