Skip to content

Commit 59ac0fb

Browse files
stidsborgstidsborg
committed
Inline completed flows' pending messages into effect state instead of parking positions
A message fetched for a completed flow is moved into the flow's reserved pending-messages effect entry (serializer-independent BinaryPacker encoding) and its row deleted - any later re-invocation, on any replica and via any restart path, receives it from the effect snapshot the restart hands over. The QueueManager stages the entry at initialization (deduped by position and idempotency key) and prunes it atomically with each delivery's capture. The effect write demands the flow is unowned (owner IS NULL) so it cannot interleave with a claim; rows are deleted only while the flow is observed still completed afterwards, otherwise the positions are reopened and normal delivery takes over. The inline uses the store's current rows, not the fetched copies, so concurrent control-panel Replace/Clear/Remove edits win. PostgreSQL now stores effects in a column on the flows table (fillfactor 80) matching MariaDB/SqlServer, gaining the owner-guarded effect write it lacked and making claim + effect snapshot a single atomic UPDATE..RETURNING; the separate effects table, its unused version machinery and the orphaned ReadMessagesForMultipleStores reader are removed. The in-memory store gains the unowned-flow guard. ExistingMessages surfaces and edits the union of both message carriers. The per-replica parked-positions state is deleted.
1 parent c8169a5 commit 59ac0fb

20 files changed

Lines changed: 605 additions & 342 deletions

File tree

‎Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/MessagingTests.cs‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,4 +33,11 @@ public override Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessa
3333
[TestMethod]
3434
public override Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel()
3535
=> EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(FunctionStoreFactory.Create());
36-
}
36+
[TestMethod]
37+
public override Task PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog()
38+
=> PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog(FunctionStoreFactory.Create());
39+
40+
[TestMethod]
41+
public override Task PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica()
42+
=> PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica(FunctionStoreFactory.Create());
43+
}

‎Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/MessagingTests.cs‎

Lines changed: 119 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -284,13 +284,125 @@ await store.MessageStore.AppendMessages([
284284
controlPanel.Status.ShouldBe(Status.Succeeded);
285285
controlPanel.Result.ShouldBe("hello world");
286286

287-
// The delivered message is deleted after delivery. The empty message is either consumed by a watchdog
288-
// restart along the way (deleted) or remains parked for the completed flow - but is never delivered.
289-
await BusyWait.Until(async () =>
290-
{
291-
var remaining = await store.MessageStore.GetMessages(storedId);
292-
return remaining.All(m => m.IsEmpty);
293-
});
287+
// Both rows end up deleted: the non-empty message is inlined into the completed flow's effect state (and
288+
// its row removed), while the empty poke is simply deleted - a completed flow needs no restart.
289+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
290+
291+
unhandledExceptionHandler.ShouldNotHaveExceptions();
292+
}
293+
294+
public abstract Task PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog();
295+
public async Task PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog(Task<IFunctionStore> functionStore)
296+
{
297+
var store = await functionStore;
298+
299+
var flowId = TestFlowId.Create();
300+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
301+
using var functionsRegistry = new FunctionsRegistry(
302+
store,
303+
new Settings(
304+
unhandledExceptionHandler.Catch,
305+
messagesPullFrequency: TimeSpan.FromMilliseconds(100),
306+
watchdogCheckFrequency: TimeSpan.FromMilliseconds(100)
307+
)
308+
);
309+
310+
var awaitMessage = new SyncedFlag();
311+
var registration = functionsRegistry.RegisterFunc(
312+
flowId.Type,
313+
inner: async Task<string> (string _, Workflow workflow) =>
314+
awaitMessage.IsRaised ? await workflow.Message<string>() : "no message awaited"
315+
);
316+
317+
// Complete the flow without it consuming any messages.
318+
await registration.Run(flowId.Instance.Value, "");
319+
320+
var storedId = registration.MapToStoredId(flowId.Instance);
321+
var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
322+
var serializer = DefaultSerializer.Instance;
323+
await store.MessageStore.AppendMessages([
324+
new StoredIdAndMessage(
325+
storedId,
326+
new StoredMessage(
327+
serializer.Serialize("hello world", typeof(string)),
328+
serializer.SerializeType(typeof(string)),
329+
Position: 0,
330+
Replica: replicaId
331+
)
332+
)
333+
]);
334+
335+
// The message is inlined into the completed flow's effect state and its row deleted.
336+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
337+
338+
// Resurrect the completed flow via Postpone - the PostponedWatchdog's restart path must also hand the
339+
// inlined message over (it travels in the effect snapshot, not through any restart-specific channel).
340+
awaitMessage.Raise();
341+
var controlPanel = await registration.ControlPanel(flowId.Instance);
342+
controlPanel.ShouldNotBeNull();
343+
await controlPanel.Postpone(DateTime.UtcNow);
344+
345+
await controlPanel.BusyWaitUntil(c => c.Status == Status.Succeeded);
346+
controlPanel.Result.ShouldBe("hello world");
347+
348+
unhandledExceptionHandler.ShouldNotHaveExceptions();
349+
}
350+
351+
public abstract Task PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica();
352+
public async Task PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica(Task<IFunctionStore> functionStore)
353+
{
354+
var store = await functionStore;
355+
356+
var flowId = TestFlowId.Create();
357+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
358+
var awaitMessage = new SyncedFlag();
359+
360+
Func<FunctionsRegistry> createRegistry = () => new FunctionsRegistry(
361+
store,
362+
new Settings(unhandledExceptionHandler.Catch, messagesPullFrequency: TimeSpan.FromMilliseconds(100))
363+
);
364+
Func<FunctionsRegistry, FuncRegistration<string, string>> register = registry => registry.RegisterFunc(
365+
flowId.Type,
366+
inner: async Task<string> (string _, Workflow workflow) =>
367+
awaitMessage.IsRaised ? await workflow.Message<string>() : "no message awaited"
368+
);
369+
370+
using var publisherRegistry = createRegistry();
371+
var publisherRegistration = register(publisherRegistry);
372+
373+
// Complete the flow on the publisher replica without it consuming any messages.
374+
await publisherRegistration.Run(flowId.Instance.Value, "");
375+
376+
// Append a message stamped with the publisher replica - its watchdog fetches it, finds the flow
377+
// completed and inlines it into the flow's effect state.
378+
var storedId = publisherRegistration.MapToStoredId(flowId.Instance);
379+
var serializer = DefaultSerializer.Instance;
380+
await store.MessageStore.AppendMessages([
381+
new StoredIdAndMessage(
382+
storedId,
383+
new StoredMessage(
384+
serializer.Serialize("hello world", typeof(string)),
385+
serializer.SerializeType(typeof(string)),
386+
Position: 0,
387+
Replica: publisherRegistry.ClusterInfo.ReplicaId
388+
)
389+
)
390+
]);
391+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
392+
393+
// Restart the flow from a DIFFERENT replica: the inlined message travels in the effect snapshot the
394+
// restart hands over, so delivery does not depend on the publisher replica's in-memory state.
395+
using var restartingRegistry = createRegistry();
396+
var restartingRegistration = register(restartingRegistry);
397+
398+
awaitMessage.Raise();
399+
var controlPanel = await restartingRegistration.ControlPanel(flowId.Instance);
400+
controlPanel.ShouldNotBeNull();
401+
await controlPanel.ScheduleRestart();
402+
await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);
403+
await controlPanel.Refresh();
404+
controlPanel.Status.ShouldBe(Status.Succeeded);
405+
controlPanel.Result.ShouldBe("hello world");
294406

295407
unhandledExceptionHandler.ShouldNotHaveExceptions();
296408
}

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs‎

Lines changed: 101 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@
66
using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
77
using Cleipnir.ResilientFunctions.Domain;
88
using Cleipnir.ResilientFunctions.Messaging;
9+
using Cleipnir.ResilientFunctions.Queuing;
910
using Cleipnir.ResilientFunctions.Storage;
11+
using Cleipnir.ResilientFunctions.Storage.Session;
1012

1113
namespace Cleipnir.ResilientFunctions.CoreRuntime;
1214

@@ -117,21 +119,14 @@ public async Task RestartExecutions(IEnumerable<StoredMessages> messages)
117119
}
118120

119121
// Flows that could not be claimed were never delivered to, yet the MessageWatchdog optimistically marked
120-
// their positions as pushed. Their positions must leave the pushed-set again: parked per flow when the
121-
// flow has completed (kept in the ignore-set until an explicit re-invocation), reopened for re-fetch
122-
// otherwise.
122+
// their positions as pushed. Completed flows can never consume their messages - inline them into the
123+
// flow's effect state (and delete the rows) so any later re-invocation, on any replica and via any
124+
// restart path, finds them in the effect snapshot the restart hands over. All other flows may become
125+
// claimable later (executing elsewhere, a lost claim race, or a flow that has not been created yet -
126+
// messages may legally precede their flow): reopen their positions so the messages are re-fetched.
123127
foreach (var (storedId, storedMessagesList) in groups.Where(kv => !results.ContainsKey(kv.Key)))
124128
{
125-
// Park BEFORE reading the status. An explicit re-invocation of a completed flow claims it and then
126-
// reopens its parked positions - so whichever side acts second sees the other's write: a park landing
127-
// before the restart's reopen is released by that reopen, while a park landing after it also lands
128-
// after the claim, making the status read below observe the no-longer-completed flow and release the
129-
// park. Checking the status first would leave a window where the restart's reopen misses a park based
130-
// on a stale completed-status read, stranding the positions in the ignore-set forever.
131-
_messageClearer.ParkPositions(
132-
storedId,
133-
storedMessagesList.SelectMany(sm => sm.Messages).Select(m => m.Position).ToList()
134-
);
129+
var flowMessages = storedMessagesList.SelectMany(sm => sm.Messages).ToList();
135130

136131
StoredFlow? storedFlow;
137132
try
@@ -140,20 +135,15 @@ public async Task RestartExecutions(IEnumerable<StoredMessages> messages)
140135
}
141136
catch
142137
{
143-
// Status unknown - release the park below so delivery is retried rather than the positions
144-
// being stranded.
138+
// Status unknown - reopen below so delivery is retried rather than the positions being stranded.
145139
storedFlow = null;
146140
}
147141

148142
if (storedFlow != null && storedFlow.Status is Status.Succeeded or Status.Failed)
149-
// Completed flows can never consume their messages - the positions stay parked (still in the
150-
// ignore-set, so no re-fetch churn) until an explicit re-invocation reopens them.
151-
continue;
152-
153-
// The flow may become claimable later (executing elsewhere, a lost claim race, or a flow that has
154-
// not been created yet - messages may legally precede their flow): release the park so the messages
155-
// are re-fetched - without deleting them from the store.
156-
_messageClearer.ReopenParkedPositions(storedId);
143+
if (await TryInlinePendingMessages(storedId, flowMessages))
144+
continue;
145+
146+
_messageClearer.ReopenPositions(flowMessages.Select(m => m.Position));
157147
}
158148

159149
// Resume each restarted flow, supplying the messages we already hold so it does not re-fetch them. Empty
@@ -188,4 +178,92 @@ public async Task RestartExecutions(IEnumerable<StoredMessages> messages)
188178
if (restartedEmptyPositions.Count > 0)
189179
await _messageClearer.Clear(restartedEmptyPositions);
190180
}
181+
182+
/// <summary>
183+
/// Persists a completed flow's in-hand messages into its effect state (the reserved pending-messages entry)
184+
/// and deletes them from the message store - empty restart-pokes are just deleted (a completed flow needs no
185+
/// restart). The effect write demands the flow is unowned (owner IS NULL), so it cannot interleave with a
186+
/// claim: a write that succeeds happened before any claim, and that claim's effect snapshot - read after the
187+
/// claim on every store - therefore includes the entry. After the verified write the status is re-read; if
188+
/// the flow has been resurrected in the meantime (e.g. it was actually suspended at write time), false is
189+
/// returned so the caller reopens the positions and normal delivery takes over - the then-redundant entry is
190+
/// erased by the incarnation's own flushes or pruned on delivery. Returns true when the messages were inlined
191+
/// and their rows deleted.
192+
/// </summary>
193+
private async Task<bool> TryInlinePendingMessages(StoredId storedId, IReadOnlyList<StoredMessage> messages)
194+
{
195+
try
196+
{
197+
// Inline from the store's CURRENT rows, not the in-hand copies: control-panel tooling may have
198+
// replaced (stale content) or deleted (Clear/Remove) rows since the fetch - a deleted row must stay
199+
// deleted and a replaced row must be inlined with its fresh content. In-hand positions whose rows are
200+
// gone are still cleared below, which trims them from the ignore-set (the row delete is a no-op).
201+
var inHandPositions = messages.Where(m => !m.IsEmpty).Select(m => m.Position).ToHashSet();
202+
var currentRows = await _functionStore.MessageStore.GetMessages(storedId);
203+
var deliverable = currentRows.Where(m => !m.IsEmpty && inHandPositions.Contains(m.Position)).ToList();
204+
205+
if (deliverable.Count > 0)
206+
{
207+
// Merge-write-verify loop: the owner guard does not serialize two concurrent unowned-flow writers
208+
// (both pass owner IS NULL), so the merged entry is re-read until this batch's messages are
209+
// observed to have survived - a lost update is simply retried.
210+
var verified = false;
211+
for (var attempt = 0; attempt < 5 && !verified; attempt++)
212+
{
213+
var effects = await _functionStore.EffectsStore.GetEffectResults(storedId);
214+
var byPosition = new Dictionary<long, StoredMessage>();
215+
var existingEntry = effects.FirstOrDefault(e => e.EffectId == PendingMessages.EffectId);
216+
if (existingEntry?.Result is { Length: > 0 } existingBytes)
217+
foreach (var pending in PendingMessages.Decode(existingBytes))
218+
byPosition[pending.Position] = pending;
219+
220+
if (deliverable.All(m => byPosition.ContainsKey(m.Position)))
221+
{
222+
verified = true;
223+
continue;
224+
}
225+
226+
foreach (var message in deliverable)
227+
byPosition[message.Position] = message;
228+
229+
var session = new SnapshotStorageSession(replicaId: null)
230+
{
231+
RowExists = true,
232+
Version = SnapshotStorageSession.NoVersionCheck
233+
};
234+
foreach (var effect in effects)
235+
session.Effects[effect.EffectId] = effect;
236+
237+
var entry = StoredEffect.CreateCompleted(
238+
PendingMessages.EffectId,
239+
PendingMessages.Encode(byPosition.Values.OrderBy(m => m.Position).ToList()),
240+
alias: null
241+
);
242+
await _functionStore.EffectsStore.SetEffectResult(
243+
storedId,
244+
new StoredEffectChange(storedId, PendingMessages.EffectId, CrudOperation.Insert, entry),
245+
session
246+
);
247+
}
248+
249+
if (!verified)
250+
return false;
251+
}
252+
253+
// Delete the rows only while the flow is still completed - otherwise keep them (caller reopens) and
254+
// let normal delivery handle them.
255+
var storedFlow = await _functionStore.GetFunction(storedId);
256+
if (storedFlow == null || storedFlow.Status is not (Status.Succeeded or Status.Failed))
257+
return false;
258+
259+
await _messageClearer.Clear(messages.Select(m => m.Position).ToList());
260+
return true;
261+
}
262+
catch
263+
{
264+
// Includes the owner-guard's concurrent-modification signal (the flow was claimed mid-write) - the
265+
// caller reopens the positions and the next poll retries.
266+
return false;
267+
}
268+
}
191269
}

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/InvocationHelper.cs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -224,10 +224,10 @@ public async Task PublishCompletionMessageToParent(StoredId? parent, FlowId chil
224224
if (restarted == null)
225225
return null;
226226

227-
// The restart does not pull the flow's messages - message fetching is the MessageWatchdog's sole
228-
// responsibility. Release any positions parked while the flow was completed and wake the watchdog, so
229-
// pending messages are fetched and pushed to the restarted flow now rather than on the next poll.
230-
_messageClearer.ReopenParkedPositions(flowId);
227+
// The restart does not pull the flow's messages: store-resident messages are fetched and pushed by the
228+
// MessageWatchdog (woken here so they arrive now rather than on the next poll), while messages inlined
229+
// into the effect state while the flow was completed travel in the effect snapshot handed over below and
230+
// are staged by the QueueManager at initialization.
231231
_messageWatchdog.Notify();
232232

233233
return new RestartedFunction(
@@ -421,7 +421,7 @@ public async Task<ExistingEffects> CreateExistingEffects(FlowId flowId)
421421
var storedEffects = await _functionStore.EffectsStore.GetEffectResults(storedId);
422422
return new ExistingEffects(storedId, flowId, _functionStore.EffectsStore, Serializer, storedEffects);
423423
}
424-
public ExistingMessages CreateExistingMessages(FlowId flowId) => new(MapToStoredId(flowId), _functionStore.MessageStore, Serializer, _replicaId);
424+
public ExistingMessages CreateExistingMessages(FlowId flowId) => new(MapToStoredId(flowId), _functionStore.MessageStore, _functionStore.EffectsStore, Serializer, _replicaId);
425425

426426
public QueueManager CreateQueueManager(FlowId flowId, StoredId storedId, Effect effect, FlowExecutionState flowExecutionState, FlowTimeouts timeouts, UnhandledExceptionHandler unhandledExceptionHandler)
427427
=> new(flowId, storedId, Serializer, effect, flowExecutionState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Watchdogs/IMessageClearer.cs‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,17 @@
11
using System.Collections.Generic;
22
using System.Threading.Tasks;
3-
using Cleipnir.ResilientFunctions.Storage;
43

54
namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
65

76
/// <summary>
87
/// The slice of <see cref="MessageClearer"/> its collaborators depend on: the QueueManager deletes handled messages
98
/// from the store (and drops their positions from the watchdog's ignore-set) via <see cref="Clear"/>, while the
109
/// FlowsManager drops the positions of flows it could not restart back out of the ignore-set - without deleting them
11-
/// from the store - via <see cref="ReopenPositions"/>. Positions of completed flows are parked per flow via
12-
/// <see cref="ParkPositions"/> (kept in the ignore-set so they are not pointlessly re-fetched) and released again by
13-
/// <see cref="ReopenParkedPositions"/> when the flow is explicitly re-invoked. Exists so tests can pass a no-op stub
14-
/// instead of a fully wired clearer.
10+
/// from the store - via <see cref="ReopenPositions"/>. Exists so tests can pass a no-op stub instead of a fully
11+
/// wired clearer.
1512
/// </summary>
1613
internal interface IMessageClearer
1714
{
1815
Task Clear(IReadOnlyList<long> positions);
1916
void ReopenPositions(IEnumerable<long> positions);
20-
void ParkPositions(StoredId storedId, IReadOnlyList<long> positions);
21-
void ReopenParkedPositions(StoredId storedId);
2217
}

0 commit comments

Comments
 (0)