Persist effect result types as ids through a dedicated type store (#254)

* Rename ITypeStore to IFlowTypeStore and its implementations

* Persist effect result types as ids through a dedicated type store

StoredEffect.ResultType is now a long - the first 8 bytes of the SHA-256
hash of the type's encoded form (UTF-8 of its simple qualified name) -
instead of the inline encoded type. The new ITypeStore persists the
id -> encoded-type mapping in a {prefix}_dotnet_types table in every
store; ids are content-derived, so inserts are idempotent.

The registry-wide TypeMapper computes ids without touching the store and
tracks which mappings are known-persisted. Every effect write path
awaits TypeMapper.EnsurePersisted before handing effects to the store
(EffectResults.Flush, control-panel effect writes, staged-message
children and CreateFunction's initial effects), so a mapping row is
always durable before the first effect referencing it - a crash between
the two writes can never leave effects whose results cannot be
deserialized. Resolution of an unknown id refreshes the whole (small)
mapping table once per process.

The flow-type store moves to IFunctionStore.FlowTypeStore, freeing the
TypeStore property for the new store. ISerializer no longer serializes
types at all: the SerializeType/ResolveType default methods are deleted
and the encoding lives in TypeHelper extensions, shared by the TypeMapper
and the message paths (where message types remain inline bytes).

* Track persistedness by insertion order instead of a separate marker set

_serializedTypes now only ever contains mappings that are durable in the
type store: an entry is added after its insert has completed (or from a
store refresh), never before - making the dictionary itself the persisted
marker and removing the separate _persisted set. Concurrent
EnsurePersisted calls may insert the same mapping more than once; inserts
are idempotent, so duplicates are accepted rather than queued behind a
lock.

Bytes for a minted-but-not-yet-persisted id are recovered by a reverse
lookup over the ids GetTypeId has handed out - both when persisting the
mapping and when an effect created in this process is read back before
its first flush.

* Persist message types as TypeIds and thread the id through the message model

TypeId - a readonly struct wrapping the content-derived long - replaces
raw longs and inline encoded types everywhere a type travels: an effect's
ResultType, SerializedMessage.Type, StoredMessage/StoredDlqMessage's
MessageType and the PendingMessages effect-carrier encoding all carry a
TypeId now (8 bytes in the packed forms). A null message type marks an
empty restart-poke - the poke's empty-content-and-type encoding is gone.

Message producers mint ids through the registry's TypeMapper and
consumers resolve through it; the ISerializer-independent inline type
bytes are gone from the message pipeline. EnsurePersisted becomes
parameterless - it persists every id minted by this process that is not
yet known durable - so ids buried inside already-encoded payloads
(staged-message children, delivered-message captures) are covered by the
effect flush without threading them around, and MessageSender persists
before appending rows for the same reason.

Test-side, DefaultDeserialize takes the mapper it resolves through, and
directly-constructed messages mint-and-persist their type id via the new
TypeIdTestHelper.

* Name the type-mapping table _types and the flow-type table _flowtypes

The .NET-type mapping is the primary type table, so it takes the _types
name; the flow-type table moves to _flowtypes (matching its
IFlowTypeStore rename on main).

* Track unpersisted type mappings directly instead of sweeping all minted ids

GetTypeId records a newly minted mapping in an unpersisted dictionary,
making EnsurePersisted's fast path an emptiness check instead of an
iteration over every minted type, and making unpersisted bytes
addressable by id - which replaces the reverse lookup over minted ids in
both the persist and the read-back-before-first-flush resolution paths.
Entries move to the persisted dictionary before they are drained, so a
concurrent reader always finds a mapping in at least one of the two.

* Cache resolved TypeId -> Type lookups

GetTypeId seeds the cache with the Type it already has in hand, and the
first resolution of a foreign id fills it lazily - so repeated
ResolveType calls are a single dictionary lookup instead of a
Type.GetType round-trip over the encoded name.

* Resolve type ids asynchronously instead of blocking on the store

TypeMapper.ResolveType returns a Task<Type>: the refresh a foreign id
falls back on was blocked on with GetAwaiter().GetResult(), and three of
its four call chains reached it from inside EffectResults' _sync lock -
so one first-resolution parked a thread pool thread on a store
round-trip and held every other effect operation on that flow behind it.
Nothing required the synchronous signature; every call site bottoms out
in an already-async method.

EffectResults' three deserializing paths (TryGet, CreateOrGet and the
generic InnerCapture) now snapshot the StoredEffect under the lock and
resolve outside it - safe because the pending change and the stored
effect are immutable records - so no store round-trip happens while the
effect state is locked. The new await sits only on the already-completed
return path, leaving the write paths' interleaving unchanged.

Out params cannot survive async, so EffectResults/Effect.TryGet return
(bool Success, T? Value) and Effect.Get returns a Task; both are
internal. IdempotencyKeys.Initialize follows its caller into async. The
public surface changes are ResolveType, the ResolveResultType extension
and the two test-only DefaultDeserialize methods, which now return tasks
- the test helper blocks on them rather than rippling awaits through
assertion sites, some of which sit inside LINQ predicates.

* Admit one type-store refresh at a time

Resolving an unknown id refreshes the whole mapping table, so concurrent
misses each fetched it in full. A restart reading back a batch of
foreign payloads misses on many distinct ids at once, which made the cost
scale with the number of unknown types rather than with the one fetch
that already covers all of them.

A semaphore admits one refresh, and waiters re-check the persisted
mappings before reaching the store: the refresh they queued behind
fetched every type, so their id is present and they return without
further I/O. Only an id that genuinely is not in the store - the path to
the TypeLoadException below - refreshes again, so a mapping written
between the two attempts is still picked up rather than cached away as a
permanent miss.

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/EffectFlushConcurrencyTests.cs

@@ -26,6 +26,7 @@ private static Effect CreateEffect(StoredId storedId, IFunctionStore functionSto
             existingEffects: new List<StoredEffect>(),
             functionStore,
             DefaultSerializer.Instance,
+            new TypeMapper(functionStore.TypeStore),
             owner: null,
             storageSession: null,
             clearChildren: true
@@ -133,7 +134,8 @@ public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect
             await inner.SetEffectResults(storedId, changes, owner, session);
         }
 
-        public IFlowTypeStore TypeStore => inner.TypeStore;
+        public IFlowTypeStore FlowTypeStore => inner.FlowTypeStore;
+        public ITypeStore TypeStore => inner.TypeStore;
         public IMessageStore MessageStore => inner.MessageStore;
         public IDlqStore DlqStore => inner.DlqStore;
         public IReplicaStore ReplicaStore => inner.ReplicaStore;

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/EffectResultTypeTests.cs

@@ -25,6 +25,7 @@ private static Effect CreateEffect(StoredId storedId, IFunctionStore functionSto
             existingEffects ?? new List<StoredEffect>(),
             functionStore,
             DefaultSerializer.Instance,
+            CreateTypeMapper(functionStore),
             owner: null,
             storageSession: null,
             clearChildren: true
@@ -38,6 +39,14 @@ private static Effect CreateEffect(StoredId storedId, IFunctionStore functionSto
         );
     }
 
+    // A fresh mapper per call: resolving a stored effect's type through it goes via the type store, verifying
+    // the id -> type mapping was actually persisted alongside the effect.
+    private static TypeMapper CreateTypeMapper(IFunctionStore functionStore)
+        => new(functionStore.TypeStore);
+
+    private static Task<Type> ResolveResultType(IFunctionStore store, StoredEffect storedEffect)
+        => storedEffect.ResolveResultType(CreateTypeMapper(store));
+
     private static async Task<StoredEffect> GetStoredEffect(IFunctionStore store, StoredId storedId, EffectId effectId)
         => (await store.GetFunction(storedId))!.Effects!.Single(e => e.EffectId == effectId);
 
@@ -55,7 +64,7 @@ public async Task CapturedResultIsPersistedWithItsType()
         await effect.Capture(() => "SomeResult".ToTask());
 
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        var resultType = DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!);
+        var resultType = await ResolveResultType(store, storedEffect);
         resultType.ShouldBe(typeof(string));
         DefaultSerializer.Instance.Deserialize(storedEffect.Result!, resultType!).ShouldBe("SomeResult");
     }
@@ -72,7 +81,7 @@ public async Task UpsertedValueIsPersistedWithItsType()
         await effect.Upsert(effectId, value: 42, alias: null, flush: true);
 
         var storedEffect = await GetStoredEffect(store, storedId, effectId);
-        var resultType = DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!);
+        var resultType = await ResolveResultType(store, storedEffect);
         resultType.ShouldBe(typeof(int));
         DefaultSerializer.Instance.Deserialize(storedEffect.Result!, resultType!).ShouldBe(42);
     }
@@ -89,7 +98,7 @@ public async Task CreateOrGetValueIsPersistedWithItsType()
         await effect.CreateOrGet(effectId, value: new Person("Peter", 32), alias: null, flush: true);
 
         var storedEffect = await GetStoredEffect(store, storedId, effectId);
-        var resultType = DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!);
+        var resultType = await ResolveResultType(store, storedEffect);
         resultType.ShouldBe(typeof(Person));
         DefaultSerializer.Instance.Deserialize(storedEffect.Result!, resultType!).ShouldBe(new Person("Peter", 32));
     }
@@ -106,7 +115,7 @@ public async Task ResultCapturedThroughBaseTypeIsPersistedAndReadBackAsItsActual
         await effect.Capture<Animal>(() => Task.FromResult<Animal>(new Dog("Fido", Breed: "Beagle")));
 
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!).ShouldBe(typeof(Dog));
+        (await ResolveResultType(store, storedEffect)).ShouldBe(typeof(Dog));
 
         // Replaying the same capture against the persisted effect returns the instance that was captured -
         // not an Animal-shaped shell of it.
@@ -133,7 +142,7 @@ public async Task LazilyTypedSequenceIsMaterializedBeforeItIsPersisted()
         await effect.Capture<IEnumerable<string>>(() => Task.FromResult(names.Where(n => n.Length == 5)));
 
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!).ShouldBe(typeof(List<string>));
+        (await ResolveResultType(store, storedEffect)).ShouldBe(typeof(List<string>));
 
         EffectContext.Reset();
         var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
@@ -157,7 +166,7 @@ public async Task LazilyTypedSequenceCapturedAsObjectIsMaterializedBeforeItIsPer
         await effect.Capture<object>(() => Task.FromResult<object>(numbers.Select(n => n * 2)));
 
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!).ShouldBe(typeof(List<int>));
+        (await ResolveResultType(store, storedEffect)).ShouldBe(typeof(List<int>));
 
         // Without the materialized type the declared type is all there is to go on, and object yields a
         // JsonElement rather than the captured sequence.
@@ -181,7 +190,7 @@ public async Task PubliclyNamedCollectionIsPersistedAsIs()
         await effect.Capture<IEnumerable<string>>(() => Task.FromResult<IEnumerable<string>>(new[] { "Peter", "Ole" }));
 
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!).ShouldBe(typeof(string[]));
+        (await ResolveResultType(store, storedEffect)).ShouldBe(typeof(string[]));
     }
 
     [TestMethod]
@@ -197,7 +206,7 @@ public async Task DictionaryIsPersistedAsIs()
         await effect.Capture<IDictionary<string, int>>(() => Task.FromResult<IDictionary<string, int>>(dictionary));
 
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!).ShouldBe(typeof(Dictionary<string, int>));
+        (await ResolveResultType(store, storedEffect)).ShouldBe(typeof(Dictionary<string, int>));
 
         EffectContext.Reset();
         var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
@@ -221,7 +230,7 @@ public async Task NonVisibleReadOnlyDictionaryIsMaterializedIntoADictionaryBefor
 
         // Materialized as a dictionary - not a list of pairs - so the payload keeps its JSON-object shape.
         var storedEffect = await GetSingleStoredEffect(store, storedId);
-        DefaultSerializer.Instance.ResolveType(storedEffect.ResultType!).ShouldBe(typeof(Dictionary<string, int>));
+        (await ResolveResultType(store, storedEffect)).ShouldBe(typeof(Dictionary<string, int>));
 
         EffectContext.Reset();
         var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/SerializationTests.cs

@@ -18,8 +18,8 @@ public void ConcreteTypeOfEventIsSerializedAndDeserializedByDefaultSerializer()
         var serializer = DefaultSerializer.Instance;
         Parent @event = new Child("Hello World");
         var content = serializer.Serialize(@event, @event.GetType());
-        var type = serializer.SerializeType(@event.GetType());
-        var deserialized = serializer.Deserialize(content, serializer.ResolveType(type)!);
+        var type = @event.GetType().SerializeType();
+        var deserialized = serializer.Deserialize(content, type.ResolveType()!);
         if (deserialized is not Child child)
             throw new Exception("Expected event to be of child-type");
 
@@ -65,30 +65,4 @@ public void ExceptionCanBeConvertedToAndFromFatalWorkflowException()
 
     public record Parent;
     public record Child(string Value) : Parent;
-    
-    [TestMethod]
-    public void ImplementingClassCanOverrideResolveTypeDefaultMethod()
-    {
-        ISerializer defaultSerializer = DefaultSerializer.Instance;
-        ISerializer customSerializer = new CustomResolveTypeSerializer();
-
-        // Default implementation uses Type.GetType
-        defaultSerializer.ResolveType(typeof(string).SimpleQualifiedName().ToUtf8Bytes()).ShouldBe(typeof(string));
-
-        // Custom implementation always returns typeof(int) regardless of input
-        customSerializer.ResolveType(typeof(string).SimpleQualifiedName().ToUtf8Bytes()).ShouldBe(typeof(int));
-        customSerializer.ResolveType("anything".ToUtf8Bytes()).ShouldBe(typeof(int));
-    }
-
-    private class CustomResolveTypeSerializer : ISerializer
-    {
-        public byte[] Serialize(object value, Type type)
-            => throw new NotImplementedException();
-
-        public object Deserialize(byte[] bytes, Type type)
-            => throw new NotImplementedException();
-
-        // Override the default interface method
-        public Type? ResolveType(byte[] type) => typeof(int);
-    }
 }

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreTests.cs

@@ -142,6 +142,10 @@ public override Task BulkScheduleWithEmptyCollectionReturnsZero()
     public override Task DifferentTypesAreFetchedByGetExpiredFunctionsCall()
         => DifferentTypesAreFetchedByGetExpiredFunctionsCall(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task FlowTypeStoreSunshineScenarioTest()
+        => FlowTypeStoreSunshineScenarioTest(FunctionStoreFactory.Create());
+
     [TestMethod]
     public override Task TypeStoreSunshineScenarioTest()
         => TypeStoreSunshineScenarioTest(FunctionStoreFactory.Create());

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoredEffectSerializationTests.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Storage;
@@ -199,11 +200,12 @@ public void StoredEffectWithEmptyIdCanBeSerializedAndDeserialized()
     }
 
     [TestMethod]
-    public void ResultTypeIsSerializedAndDeserializedAlongsideResult()
+    public async Task ResultTypeIsSerializedAndDeserializedAlongsideResult()
     {
         var effectId = new EffectId([1]);
         var result = DefaultSerializer.Instance.Serialize("SomeResult", typeof(string));
-        var resultType = DefaultSerializer.Instance.SerializeType(typeof(string));
+        var typeMapper = new TypeMapper(new InMemoryTypeStore());
+        var resultType = typeMapper.GetTypeId(typeof(string));
         var storedEffect = StoredEffect.CreateCompleted(effectId, result, resultType, alias: null);
 
         var serialized = storedEffect.Serialize();
@@ -212,7 +214,7 @@ public void ResultTypeIsSerializedAndDeserializedAlongsideResult()
         deserialized.Result.ShouldBe(result);
         deserialized.ResultType.ShouldBe(resultType);
         DefaultSerializer.Instance
-            .Deserialize(deserialized.Result!, DefaultSerializer.Instance.ResolveType(deserialized.ResultType!)!)
+            .Deserialize(deserialized.Result!, await typeMapper.ResolveType(deserialized.ResultType!.Value))
             .ShouldBe("SomeResult");
     }
 

Core/Cleipnir.ResilientFunctions.Tests/Messaging/InMemoryTests/MessageClearerTests.cs

@@ -137,7 +137,7 @@ await messageStore.AppendMessages([
     }
 
     private static StoredMessage Message(StoredId storedId)
-        => new(storedId, MessageContent: new byte[] { 1 }, MessageType: new byte[] { 2 }, Position: 0, Replica: ReplicaId.Empty);
+        => new(storedId, MessageContent: new byte[] { 1 }, MessageType: new TypeId(2), Position: 0, Replica: ReplicaId.Empty);
 
     private static MessageClearer CreateClearer(
         IMessageStore messageStore,

Core/Cleipnir.ResilientFunctions.Tests/Messaging/TestTemplates/DlqManagerTests.cs

@@ -47,10 +47,10 @@ protected async Task DeadLetteredMessagesCanBeRedriven(Task<IFunctionStore> func
         var untouchedId = rFunc.MapToStoredId("untouched".ToFlowInstance());
 
         await functionStore.DlqStore.Append([
-            CreateMessage(byPositionId, "byPositionMsg"),
-            CreateMessage(byStoredIdId, "byStoredIdMsg"),
-            CreateMessage(byFlowIdId, "byFlowIdMsg"),
-            CreateMessage(untouchedId, "untouchedMsg")
+            CreateMessage(functionStore, byPositionId, "byPositionMsg"),
+            CreateMessage(functionStore, byStoredIdId, "byStoredIdMsg"),
+            CreateMessage(functionStore, byFlowIdId, "byFlowIdMsg"),
+            CreateMessage(functionStore, untouchedId, "untouchedMsg")
         ]);
 
         var dlq = functionsRegistry.DeadLetterQueue;
@@ -67,7 +67,7 @@ await functionStore.DlqStore.Append([
 
         // Only the redriven messages leave the dead letter queue.
         (await dlq.GetMessages([byPositionId, byStoredIdId, byFlowIdId])).ShouldBeEmpty();
-        (await dlq.GetMessages([untouchedId])).Single().DefaultDeserialize().ShouldBe("untouchedMsg");
+        (await dlq.GetMessages([untouchedId])).Single().DefaultDeserialize(functionStore).ShouldBe("untouchedMsg");
 
         unhandledExceptionCatcher.ShouldNotHaveExceptions();
     }
@@ -99,8 +99,8 @@ protected async Task AllDeadLetteredMessagesForFlowAreRedriven(Task<IFunctionSto
         var storedId = rFunc.MapToStoredId("instanceId".ToFlowInstance());
 
         await functionStore.DlqStore.Append([
-            CreateMessage(storedId, "first"),
-            CreateMessage(storedId, "second")
+            CreateMessage(functionStore, storedId, "first"),
+            CreateMessage(functionStore, storedId, "second")
         ]);
 
         var dlq = functionsRegistry.DeadLetterQueue;
@@ -130,7 +130,7 @@ protected async Task RedrivenMessageRetainsIdempotencyKeySenderAndReceiver(Task<
         // leaves the appended row in place for inspection.
         var storedId = TestStoredId.Create();
         await functionStore.DlqStore.Append([
-            CreateMessage(storedId, "hello world", idempotencyKey: "idempotencyKey1", sender: "sender1", receiver: "receiver1")
+            CreateMessage(functionStore, storedId, "hello world", idempotencyKey: "idempotencyKey1", sender: "sender1", receiver: "receiver1")
         ]);
 
         var dlq = functionsRegistry.DeadLetterQueue;
@@ -139,7 +139,7 @@ await functionStore.DlqStore.Append([
         var messages = await functionStore.MessageStore.GetMessages(storedId);
         messages.Count.ShouldBe(1);
         var message = messages.Single();
-        message.DefaultDeserialize().ShouldBe("hello world");
+        message.DefaultDeserialize(functionStore).ShouldBe("hello world");
         message.IdempotencyKey.ShouldBe("idempotencyKey1");
         message.Sender.ShouldBe("sender1");
         message.Receiver.ShouldBe("receiver1");
@@ -173,8 +173,8 @@ protected async Task DeletedDeadLetteredMessagesAreNotRedelivered(Task<IFunction
         var otherId = rFunc.MapToStoredId("otherInstance".ToFlowInstance());
 
         await functionStore.DlqStore.Append([
-            CreateMessage(storedId, "deleted"),
-            CreateMessage(otherId, "kept")
+            CreateMessage(functionStore, storedId, "deleted"),
+            CreateMessage(functionStore, otherId, "kept")
         ]);
 
         var dlq = functionsRegistry.DeadLetterQueue;
@@ -183,7 +183,7 @@ await functionStore.DlqStore.Append([
 
         // Unlike redrive, delete must not put the message back into the message store - the flow stays waiting.
         (await dlq.GetMessages([storedId])).ShouldBeEmpty();
-        (await dlq.GetMessages([otherId])).Single().DefaultDeserialize().ShouldBe("kept");
+        (await dlq.GetMessages([otherId])).Single().DefaultDeserialize(functionStore).ShouldBe("kept");
         (await functionStore.MessageStore.GetMessages(storedId)).ShouldBeEmpty();
         await Should.ThrowAsync<TimeoutException>(() => scheduled.Completion(timeout: TimeSpan.FromSeconds(1)));
 
@@ -205,7 +205,7 @@ protected async Task DeadLetteredMessagesCanBePagedThrough(Task<IFunctionStore>
         await functionStore.DlqStore.Append(
             Enumerable
                 .Range(0, 5)
-                .Select(i => CreateMessage(storedId, $"msg{i}"))
+                .Select(i => CreateMessage(functionStore, storedId, $"msg{i}"))
                 .ToList()
         );
 
@@ -214,18 +214,18 @@ await functionStore.DlqStore.Append(
 
         var firstPage = await dlq.GetMessages(limit: 2);
         firstPage.Count.ShouldBe(2);
-        firstPage[0].DefaultDeserialize().ShouldBe("msg0");
-        firstPage[1].DefaultDeserialize().ShouldBe("msg1");
+        firstPage[0].DefaultDeserialize(functionStore).ShouldBe("msg0");
+        firstPage[1].DefaultDeserialize(functionStore).ShouldBe("msg1");
 
         //the offset is exclusive - paging is done by passing the last returned position as the next offset
         var secondPage = await dlq.GetMessages(offset: firstPage[1].Position, limit: 2);
         secondPage.Count.ShouldBe(2);
-        secondPage[0].DefaultDeserialize().ShouldBe("msg2");
-        secondPage[1].DefaultDeserialize().ShouldBe("msg3");
+        secondPage[0].DefaultDeserialize(functionStore).ShouldBe("msg2");
+        secondPage[1].DefaultDeserialize(functionStore).ShouldBe("msg3");
 
         var thirdPage = await dlq.GetMessages(offset: secondPage[1].Position, limit: 2);
         thirdPage.Count.ShouldBe(1);
-        thirdPage[0].DefaultDeserialize().ShouldBe("msg4");
+        thirdPage[0].DefaultDeserialize(functionStore).ShouldBe("msg4");
 
         (await dlq.GetMessages(offset: thirdPage[0].Position)).ShouldBeEmpty();
 
@@ -244,7 +244,7 @@ protected async Task RedrivingAndDeletingUnknownOrEmptyInputIsANoOp(Task<IFuncti
         );
 
         var storedId = TestStoredId.Create();
-        await functionStore.DlqStore.Append([CreateMessage(storedId, "untouched")]);
+        await functionStore.DlqStore.Append([CreateMessage(functionStore, storedId, "untouched")]);
 
         var dlq = functionsRegistry.DeadLetterQueue;
 
@@ -258,16 +258,16 @@ protected async Task RedrivingAndDeletingUnknownOrEmptyInputIsANoOp(Task<IFuncti
         await dlq.Delete(new List<long>());
         await dlq.Delete([999_999L]);
 
-        (await dlq.GetMessages()).Single().DefaultDeserialize().ShouldBe("untouched");
+        (await dlq.GetMessages()).Single().DefaultDeserialize(functionStore).ShouldBe("untouched");
 
         unhandledExceptionCatcher.ShouldNotHaveExceptions();
     }
 
-    private static StoredMessage CreateMessage(StoredId storedId, string content, string? idempotencyKey = null, string? sender = null, string? receiver = null)
+    private static StoredMessage CreateMessage(IFunctionStore functionStore, StoredId storedId, string content, string? idempotencyKey = null, string? sender = null, string? receiver = null)
         => new(
             storedId,
             content.ToJson().ToUtf8Bytes(),
-            typeof(string).SimpleQualifiedName().ToUtf8Bytes(),
+            functionStore.GetTypeId(typeof(string)),
             Position: 0,
             Replica: ReplicaId.Empty,
             idempotencyKey,