Skip to content

Commit ac9929b

Browse files
committed
Add Effect.FlushlessCreateNextChild for appending numbered child effects
Introduces a flushless primitive that finds the next free direct-child slot under a given parent EffectId (highest existing child index + 1) and records the provided content there as a completed effect, returning the new child's EffectId. Index-selection and reservation happen under a single lock so concurrent callers cannot collide on the same index. Generalizes the hand-rolled next-child bookkeeping in IdempotencyKeys.
1 parent 968d3c1 commit ac9929b

3 files changed

Lines changed: 143 additions & 1 deletion

File tree

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Threading.Tasks;
4+
using Cleipnir.ResilientFunctions.CoreRuntime;
5+
using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
6+
using Cleipnir.ResilientFunctions.Domain;
7+
using Cleipnir.ResilientFunctions.Helpers;
8+
using Cleipnir.ResilientFunctions.Storage;
9+
using Cleipnir.ResilientFunctions.Tests.Utils;
10+
using Microsoft.VisualStudio.TestTools.UnitTesting;
11+
using Shouldly;
12+
13+
namespace Cleipnir.ResilientFunctions.Tests.InMemoryTests;
14+
15+
[TestClass]
16+
public class CreateNextChildTests
17+
{
18+
private static Effect CreateEffect(IReadOnlyList<StoredEffect> existingEffects)
19+
{
20+
var storedId = TestStoredId.Create();
21+
var effectResults = new EffectResults(
22+
TestFlowId.Create(),
23+
storedId,
24+
existingEffects,
25+
new InMemoryFunctionStore().EffectsStore,
26+
DefaultSerializer.Instance,
27+
storageSession: null,
28+
clearChildren: true
29+
);
30+
31+
return new Effect(
32+
effectResults,
33+
utcNow: () => DateTime.UtcNow,
34+
new FlowTimeouts(),
35+
new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance)
36+
);
37+
}
38+
39+
[TestMethod]
40+
public void CreateNextChildAppendsSequentiallyFromZero()
41+
{
42+
var effect = CreateEffect(existingEffects: new List<StoredEffect>());
43+
var parent = new EffectId([1]);
44+
45+
var first = effect.FlushlessCreateNextChild(parent, content: "a");
46+
var second = effect.FlushlessCreateNextChild(parent, content: "b");
47+
var third = effect.FlushlessCreateNextChild(parent, content: "c");
48+
49+
first.ShouldBe(parent.CreateChild(0));
50+
second.ShouldBe(parent.CreateChild(1));
51+
third.ShouldBe(parent.CreateChild(2));
52+
53+
// Flushless writes are visible in-memory immediately (no Flush required).
54+
effect.Get<string>(first).ShouldBe("a");
55+
effect.Get<string>(second).ShouldBe("b");
56+
effect.Get<string>(third).ShouldBe("c");
57+
}
58+
59+
[TestMethod]
60+
public void CreateNextChildContinuesFromExistingChildren()
61+
{
62+
var parent = new EffectId([7]);
63+
var existing = new List<StoredEffect>
64+
{
65+
StoredEffect.CreateCompleted(parent.CreateChild(0), alias: null),
66+
StoredEffect.CreateCompleted(parent.CreateChild(1), alias: null),
67+
};
68+
var effect = CreateEffect(existing);
69+
70+
var next = effect.FlushlessCreateNextChild(parent, content: "x");
71+
72+
next.ShouldBe(parent.CreateChild(2));
73+
}
74+
75+
[TestMethod]
76+
public async Task CreateNextChildKeepsGapAndUsesHighestPlusOne()
77+
{
78+
var effect = CreateEffect(existingEffects: new List<StoredEffect>());
79+
var parent = new EffectId([1]);
80+
81+
effect.FlushlessCreateNextChild(parent, content: "a"); // [1,0]
82+
var middle = effect.FlushlessCreateNextChild(parent, content: "b"); // [1,1]
83+
effect.FlushlessCreateNextChild(parent, content: "c"); // [1,2]
84+
85+
await effect.Clear(middle, flush: true); // remove [1,1], leaving {0,2}
86+
87+
var next = effect.FlushlessCreateNextChild(parent, content: "d");
88+
89+
// Highest existing index (2) + 1 -> gap at index 1 is kept.
90+
next.ShouldBe(parent.CreateChild(3));
91+
}
92+
93+
[TestMethod]
94+
public void CreateNextChildIsScopedToItsParent()
95+
{
96+
var effect = CreateEffect(existingEffects: new List<StoredEffect>());
97+
var parentA = new EffectId([1]);
98+
var parentB = new EffectId([2]);
99+
100+
var a0 = effect.FlushlessCreateNextChild(parentA, content: "a0");
101+
var a1 = effect.FlushlessCreateNextChild(parentA, content: "a1");
102+
var b0 = effect.FlushlessCreateNextChild(parentB, content: "b0");
103+
104+
a0.ShouldBe(parentA.CreateChild(0));
105+
a1.ShouldBe(parentA.CreateChild(1));
106+
b0.ShouldBe(parentB.CreateChild(0));
107+
}
108+
}

Core/Cleipnir.ResilientFunctions/Domain/Effect.cs

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,15 @@ internal T Get<T>(EffectId effectId)
117117

118118
internal IReadOnlyList<EffectId> GetChildren(EffectId effectId) => effectResults.GetChildren(effectId);
119119

120+
/// <summary>
121+
/// Finds the next free child slot under <paramref name="parentId"/> (highest existing direct-child index + 1)
122+
/// and records <paramref name="content"/> there as a completed effect, returning the created child's EffectId.
123+
/// The write is flushless (in-memory only) - call <see cref="Flush"/> to persist. This is a raw append: each
124+
/// call adds a new child, so it is not replay-idempotent on its own.
125+
/// </summary>
126+
public EffectId FlushlessCreateNextChild<T>(EffectId parentId, T content, string? alias = null)
127+
=> effectResults.FlushlessCreateNextChild(parentId, content, alias);
128+
120129
#region Implicit ids
121130

122131
public Task Capture(Action work, ResiliencyLevel resiliency = ResiliencyLevel.AtLeastOnce)

Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,26 @@ public IReadOnlyList<EffectId> GetChildren(EffectId parentId)
231231
.Where(id => id.IsDescendant(parentId))
232232
.ToList();
233233
}
234-
234+
235+
public EffectId FlushlessCreateNextChild<T>(EffectId parentId, T value, string? alias)
236+
{
237+
lock (_sync)
238+
{
239+
// Highest existing direct-child index + 1 (append). Index-selection and reservation
240+
// happen under a single lock so concurrent callers cannot pick the same index.
241+
var nextIndex = 0;
242+
foreach (var id in _effectResults.Keys)
243+
if (parentId.IsChild(id) && id.Id >= nextIndex)
244+
nextIndex = id.Id + 1;
245+
246+
var childId = parentId.CreateChild(nextIndex);
247+
var serializedValue = _serializer.Serialize(value!, typeof(T));
248+
var storedEffect = StoredEffect.CreateCompleted(childId, serializedValue, alias);
249+
AddToPendingCore(childId, storedEffect, delete: false, clearChildren: false);
250+
return childId;
251+
}
252+
}
253+
235254
public async Task InnerCapture(EffectId effectId, string? alias, Func<Task> work, ResiliencyLevel resiliency, EffectContext effectContext)
236255
{
237256
EffectContext.SetParent(effectId);
@@ -417,6 +436,12 @@ private void AddToPending(IEnumerable<StoredEffect> storedEffects)
417436
private void AddToPending(EffectId effectId, StoredEffect? storedEffect, bool delete, bool clearChildren)
418437
{
419438
lock (_sync)
439+
AddToPendingCore(effectId, storedEffect, delete, clearChildren);
440+
}
441+
442+
// Assumes _sync is already held by the caller.
443+
private void AddToPendingCore(EffectId effectId, StoredEffect? storedEffect, bool delete, bool clearChildren)
444+
{
420445
{
421446
if (_effectResults.ContainsKey(effectId))
422447
{

0 commit comments

Comments
 (0)