Skip to content

Commit 0fa440b

Browse files
stidsborgstidsborg
committed
Remove IMessageStore.AppendMessage; route all appends through AppendMessages
Collapses the message store to a single append entry point. The two callers that consumed AppendMessage's returned ReplicaId (MessageWriter and InvocationHelper.PublishCompletionMessageToParent) used it only to decide whether to interrupt/schedule the target flow. AppendMessages already interrupts every distinct target inside its batch, so the interrupt is now unconditional - which matches the documented intent that the interrupt is the suspend-race guard and watchdog backstop "even when the target is owned by another replica". This removes the now-dead IFlowsManager dependency from MessageWriter, MessageWriters and InvocationHelper (and the corresponding FunctionsRegistry wiring). The three SQL store implementations drop their duplicate single-row INSERT; InMemoryFunctionStore inlines the per-message append into AppendMessages. A test-only AppendMessage extension forwards to AppendMessages so existing message-store tests keep exercising the single-append path.
1 parent 151de33 commit 0fa440b

12 files changed

Lines changed: 46 additions & 127 deletions

File tree

‎Core/Cleipnir.ResilientFunctions.Tests/Messaging/MessageStoreTestExtensions.cs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
using System.Threading.Tasks;
2+
using Cleipnir.ResilientFunctions.Storage;
3+
4+
namespace Cleipnir.ResilientFunctions.Messaging;
5+
6+
public static class MessageStoreTestExtensions
7+
{
8+
/// <summary>
9+
/// Test convenience for appending a single message - forwards to <see cref="IMessageStore.AppendMessages"/>.
10+
/// </summary>
11+
public static Task AppendMessage(this IMessageStore messageStore, StoredId storedId, StoredMessage storedMessage)
12+
=> messageStore.AppendMessages([new StoredIdAndMessage(storedId, storedMessage)]);
13+
}

‎Core/Cleipnir.ResilientFunctions.Tests/Messaging/TestTemplates/MessagesTests.cs‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -361,16 +361,11 @@ protected async Task SenderIsPersistedAndCanBeFetched(Task<IFunctionStore> funct
361361
var serializer = DefaultSerializer.Instance;
362362
var messageStore = functionStore.MessageStore;
363363

364-
var messageWriter = new MessageWriter(storedId, messageStore, serializer, ReplicaId.NewId(), new NoOpFlowsManager());
364+
var messageWriter = new MessageWriter(storedId, messageStore, serializer, ReplicaId.NewId());
365365
await messageWriter.AppendMessage("hello world", idempotencyKey: "key1", sender: "TestSender");
366366

367367
var messages = await messageStore.GetMessages(storedId);
368368
messages.Count.ShouldBe(1);
369369
messages[0].Sender.ShouldBe("TestSender");
370370
}
371-
372-
private sealed class NoOpFlowsManager : IFlowsManager
373-
{
374-
public Task Schedule(StoredId storedId) => Task.CompletedTask;
375-
}
376371
}

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

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,12 @@ internal class InvocationHelper<TParam, TReturn>
2424
private readonly FlowType _flowType;
2525
private readonly StoredType _storedType;
2626
private readonly ReplicaId _replicaId;
27-
private readonly IFlowsManager _flowsManager;
2827
private readonly ResultBusyWaiter<TReturn> _resultBusyWaiter;
2928
public UtcNow UtcNow { get; }
3029

3130
private ISerializer Serializer { get; }
3231

33-
public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId replicaId, bool isParamlessFunction, SettingsWithDefaults settings, IFunctionStore functionStore, ShutdownCoordinator shutdownCoordinator, ISerializer serializer, UtcNow utcNow, bool clearChildren, IFlowsManager flowsManager)
32+
public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId replicaId, bool isParamlessFunction, SettingsWithDefaults settings, IFunctionStore functionStore, ShutdownCoordinator shutdownCoordinator, ISerializer serializer, UtcNow utcNow, bool clearChildren)
3433
{
3534
_flowType = flowType;
3635
_isParamlessFunction = isParamlessFunction;
@@ -43,7 +42,6 @@ public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId repl
4342
_storedType = storedType;
4443
_replicaId = replicaId;
4544
_functionStore = functionStore;
46-
_flowsManager = flowsManager;
4745
_resultBusyWaiter = new ResultBusyWaiter<TReturn>(_functionStore, Serializer);
4846
}
4947

@@ -206,9 +204,7 @@ public async Task PublishCompletionMessageToParent(StoredId? parent, FlowId chil
206204
var content = Serializer.Serialize(msg, msg.GetType());
207205
var type = Serializer.SerializeType(msg.GetType());
208206
var storedMessage = new StoredMessage(content, type, Position: 0, IdempotencyKey: $"FlowCompleted:{childId}", Replica: _replicaId);
209-
var writtenReplica = await _functionStore.MessageStore.AppendMessage(parent, storedMessage);
210-
if (writtenReplica == _replicaId)
211-
await _functionStore.Interrupt(parent);
207+
await _functionStore.MessageStore.AppendMessages([new StoredIdAndMessage(parent, storedMessage)]);
212208
}
213209

214210
public async Task<RestartedFunction?> RestartFunction(StoredId flowId)
@@ -385,7 +381,7 @@ await _functionStore.BulkScheduleFunctions(
385381
}
386382

387383
public MessageWriter CreateMessageWriter(StoredId storedId)
388-
=> new MessageWriter(storedId, _functionStore.MessageStore, Serializer, _replicaId, _flowsManager);
384+
=> new MessageWriter(storedId, _functionStore.MessageStore, Serializer, _replicaId);
389385

390386
public Effect CreateEffect(StoredId storedId, FlowId flowId, IReadOnlyList<StoredEffect> storedEffects, FlowTimeouts flowTimeouts, IStorageSession? storageSession, FlowState flowState)
391387
{

‎Core/Cleipnir.ResilientFunctions/Domain/ExistingMessages.cs‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,8 @@ public async Task Append<T>(T message, string? idempotencyKey = null) where T :
5555
{
5656
var json = _serializer.Serialize(message, message.GetType());
5757
var type = _serializer.SerializeType(message.GetType());
58-
await _messageStore.AppendMessage(
59-
_storedId, new StoredMessage(json, type, Position: 0, Replica: ReplicaId.Empty, IdempotencyKey: idempotencyKey)
60-
);
58+
var storedMessage = new StoredMessage(json, type, Position: 0, Replica: ReplicaId.Empty, IdempotencyKey: idempotencyKey);
59+
await _messageStore.AppendMessages([new StoredIdAndMessage(_storedId, storedMessage)]);
6160

6261
// Invalidate cache so it will be re-fetched with correct positions
6362
_receivedMessages = null;

‎Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs‎

Lines changed: 6 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -246,8 +246,7 @@ public FuncRegistration<TParam, TReturn> RegisterFunc<TParam, TReturn>(
246246
_shutdownCoordinator,
247247
serializer,
248248
_settings.UtcNow,
249-
settings?.ClearChildrenAfterCapture ?? true,
250-
_flowsManager
249+
settings?.ClearChildrenAfterCapture ?? true
251250
);
252251
var invoker = new Invoker<TParam, TReturn>(
253252
flowType,
@@ -283,8 +282,7 @@ public FuncRegistration<TParam, TReturn> RegisterFunc<TParam, TReturn>(
283282
storedType,
284283
_functionStore,
285284
serializer,
286-
ClusterInfo.ReplicaId,
287-
_flowsManager
285+
ClusterInfo.ReplicaId
288286
);
289287

290288
var registration = new FuncRegistration<TParam, TReturn>(
@@ -331,8 +329,7 @@ private ParamlessRegistration RegisterParamless(
331329
_shutdownCoordinator,
332330
serializer,
333331
_settings.UtcNow,
334-
settings?.ClearChildrenAfterCapture ?? true,
335-
_flowsManager
332+
settings?.ClearChildrenAfterCapture ?? true
336333
);
337334
var invoker = new Invoker<Unit, Unit>(
338335
flowType,
@@ -368,8 +365,7 @@ private ParamlessRegistration RegisterParamless(
368365
storedType,
369366
_functionStore,
370367
serializer,
371-
ClusterInfo.ReplicaId,
372-
_flowsManager
368+
ClusterInfo.ReplicaId
373369
);
374370

375371
var registration = new ParamlessRegistration(
@@ -416,8 +412,7 @@ public ActionRegistration<TParam> RegisterAction<TParam>(
416412
_shutdownCoordinator,
417413
serializer,
418414
_settings.UtcNow,
419-
settings?.ClearChildrenAfterCapture ?? true,
420-
_flowsManager
415+
settings?.ClearChildrenAfterCapture ?? true
421416
);
422417
var rActionInvoker = new Invoker<TParam, Unit>(
423418
flowType,
@@ -453,8 +448,7 @@ public ActionRegistration<TParam> RegisterAction<TParam>(
453448
storedType,
454449
_functionStore,
455450
serializer,
456-
ClusterInfo.ReplicaId,
457-
_flowsManager
451+
ClusterInfo.ReplicaId
458452
);
459453
var registration = new ActionRegistration<TParam>(
460454
flowType,

‎Core/Cleipnir.ResilientFunctions/Messaging/IMessageStore.cs‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,12 @@ public interface IMessageStore
1010
Task Initialize();
1111

1212
/// <summary>
13-
/// Appends a message to the target flow and returns the replica written to the message row
14-
/// (the target flow's current owner, or the publishing replica when the target is not executing).
15-
/// The target flow is scheduled to run immediately when it is suspended/postponed.
13+
/// Appends the messages to their target flows and interrupts each distinct target so it runs
14+
/// immediately and consumes the message (the interrupt is the suspend-race guard and watchdog
15+
/// backstop, so the message is never lost even when the target suspends concurrently or is owned
16+
/// by another replica). Each message row is written with the target flow's current owner, or the
17+
/// publishing replica when the target is not executing.
1618
/// </summary>
17-
Task<ReplicaId> AppendMessage(StoredId storedId, StoredMessage storedMessage);
1819
Task AppendMessages(IReadOnlyList<StoredIdAndMessage> messages);
1920

2021
Task<bool> ReplaceMessage(StoredId storedId, long position, StoredMessage storedMessage);

‎Core/Cleipnir.ResilientFunctions/Messaging/MessageWriter.cs‎

Lines changed: 3 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,18 @@
11
using System.Threading.Tasks;
2-
using Cleipnir.ResilientFunctions.CoreRuntime;
32
using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
43
using Cleipnir.ResilientFunctions.Domain;
54
using Cleipnir.ResilientFunctions.Storage;
65

76
namespace Cleipnir.ResilientFunctions.Messaging;
87

9-
public class MessageWriter(StoredId storedIdId, IMessageStore messageStore, ISerializer eventSerializer, ReplicaId publisherReplica, IFlowsManager flowsManager)
8+
public class MessageWriter(StoredId storedIdId, IMessageStore messageStore, ISerializer eventSerializer, ReplicaId publisherReplica)
109
{
1110
public async Task AppendMessage<TMessage>(TMessage message, string? idempotencyKey = null, string? sender = null, string? receiver = null) where TMessage : class
1211
{
1312
var eventJson = eventSerializer.Serialize(message, message.GetType());
1413
var eventType = eventSerializer.SerializeType(message.GetType());
1514

16-
var writtenReplica = await messageStore.AppendMessage(
17-
storedIdId,
18-
new StoredMessage(eventJson, eventType, Position: 0, Replica: publisherReplica, IdempotencyKey: idempotencyKey, Sender: sender, Receiver: receiver)
19-
);
20-
21-
// The message fell back to (or is owned by) this replica - schedule the target so it runs and consumes
22-
// the message. Targets owned by another replica are delivered by that replica's MessageWatchdog.
23-
if (writtenReplica == publisherReplica)
24-
await flowsManager.Schedule(storedIdId);
15+
var storedMessage = new StoredMessage(eventJson, eventType, Position: 0, Replica: publisherReplica, IdempotencyKey: idempotencyKey, Sender: sender, Receiver: receiver);
16+
await messageStore.AppendMessages([new StoredIdAndMessage(storedIdId, storedMessage)]);
2517
}
2618
}

‎Core/Cleipnir.ResilientFunctions/Messaging/MessageWriters.cs‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,31 +14,28 @@ public class MessageWriters
1414
private readonly IFunctionStore _functionStore;
1515
private readonly ISerializer _serializer;
1616
private readonly ReplicaId _publisherReplica;
17-
private readonly IFlowsManager _flowsManager;
1817

1918
public MessageWriters(
2019
StoredType storedType,
2120
IFunctionStore functionStore,
2221
ISerializer serializer,
23-
ReplicaId publisherReplica,
24-
IFlowsManager flowsManager)
22+
ReplicaId publisherReplica)
2523
{
2624
_storedType = storedType;
2725
_functionStore = functionStore;
2826
_serializer = serializer;
2927
_publisherReplica = publisherReplica;
30-
_flowsManager = flowsManager;
3128
}
3229

3330
public MessageWriter For(FlowInstance instance)
3431
{
3532
var storedId = StoredId.Create(_storedType, instance.Value);
36-
return new MessageWriter(storedId, _functionStore.MessageStore, _serializer, _publisherReplica, _flowsManager);
33+
return new MessageWriter(storedId, _functionStore.MessageStore, _serializer, _publisherReplica);
3734
}
3835

3936
internal MessageWriter For(StoredId storedId)
4037
{
41-
return new MessageWriter(storedId, _functionStore.MessageStore, _serializer, _publisherReplica, _flowsManager);
38+
return new MessageWriter(storedId, _functionStore.MessageStore, _serializer, _publisherReplica);
4239
}
4340

4441
public async Task AppendMessages(IReadOnlyList<BatchedMessage> messages)

‎Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs‎

Lines changed: 10 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -581,27 +581,20 @@ private class InnerState
581581

582582
#region MessageStore
583583

584-
public virtual Task<ReplicaId> AppendMessage(StoredId storedId, StoredMessage storedMessage)
585-
{
586-
lock (_sync)
587-
{
588-
if (!_messages.ContainsKey(storedId))
589-
_messages[storedId] = new Dictionary<long, StoredMessage>();
590-
591-
var flowOwner = _states.TryGetValue(storedId, out var state) ? state.Owner : null;
592-
var replica = flowOwner ?? storedMessage.Replica;
593-
var messages = _messages[storedId];
594-
messages[_nextMessagePosition++] = storedMessage with { Replica = replica };
595-
596-
return Task.FromResult(replica);
597-
}
598-
}
599-
600584
public async Task AppendMessages(IReadOnlyList<StoredIdAndMessage> messages)
601585
{
602586
foreach (var (storedId, storedMessage) in messages)
603587
{
604-
await AppendMessage(storedId, storedMessage);
588+
lock (_sync)
589+
{
590+
if (!_messages.TryGetValue(storedId, out var flowMessages))
591+
flowMessages = _messages[storedId] = new Dictionary<long, StoredMessage>();
592+
593+
var flowOwner = _states.TryGetValue(storedId, out var state) ? state.Owner : null;
594+
var replica = flowOwner ?? storedMessage.Replica;
595+
flowMessages[_nextMessagePosition++] = storedMessage with { Replica = replica };
596+
}
597+
605598
await Interrupt(storedId);
606599
}
607600
}

‎Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbMessageStore.cs‎

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -48,25 +48,6 @@ public async Task TruncateTable()
4848
}
4949

5050

51-
public async Task<ReplicaId> AppendMessage(StoredId storedId, StoredMessage storedMessage)
52-
{
53-
var (messageContent, messageType, _, replica, idempotencyKey, sender, receiver) = storedMessage;
54-
var content = BinaryPacker.Pack(messageContent, messageType, idempotencyKey?.ToUtf8Bytes(), sender?.ToUtf8Bytes(), receiver?.ToUtf8Bytes());
55-
56-
var sql = @$"
57-
INSERT INTO {_tablePrefix}_messages (id, replica, content)
58-
VALUES (?, COALESCE((SELECT owner FROM {_tablePrefix} WHERE id = ?), ?), ?)
59-
RETURNING replica;";
60-
61-
await using var conn = await DatabaseHelper.CreateOpenConnection(_connectionString);
62-
await using var command = new MySqlCommand(sql, conn);
63-
command.Parameters.Add(new() { Value = storedId.AsGuid.ToString("N") });
64-
command.Parameters.Add(new() { Value = storedId.AsGuid.ToString("N") });
65-
command.Parameters.Add(new() { Value = replica.AsGuid.ToString("N") });
66-
command.Parameters.Add(new() { Value = content });
67-
return ((string) (await command.ExecuteScalarAsync())!).ParseToReplicaId();
68-
}
69-
7051
public async Task AppendMessages(IReadOnlyList<StoredIdAndMessage> messages)
7152
{
7253
if (messages.Count == 0)

0 commit comments

Comments
 (0)