Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Cleipnir.ResilientFunctions.CoreRuntime;
using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
using Cleipnir.ResilientFunctions.Domain;
using Cleipnir.ResilientFunctions.Helpers;
using Cleipnir.ResilientFunctions.Storage;
using Cleipnir.ResilientFunctions.Tests.Utils;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Shouldly;

namespace Cleipnir.ResilientFunctions.Tests.InMemoryTests;

[TestClass]
public class CreateNextChildTests
{
private static Effect CreateEffect(IReadOnlyList<StoredEffect> existingEffects)
{
var storedId = TestStoredId.Create();
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
existingEffects,
new InMemoryFunctionStore().EffectsStore,
DefaultSerializer.Instance,
storageSession: null,
clearChildren: true
);

return new Effect(
effectResults,
utcNow: () => DateTime.UtcNow,
new FlowTimeouts(),
new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance)
);
}

[TestMethod]
public void CreateNextChildAppendsSequentiallyFromZero()
{
var effect = CreateEffect(existingEffects: new List<StoredEffect>());
var parent = new EffectId([1]);

var first = effect.FlushlessCreateNextChild(parent, content: "a");
var second = effect.FlushlessCreateNextChild(parent, content: "b");
var third = effect.FlushlessCreateNextChild(parent, content: "c");

first.ShouldBe(parent.CreateChild(0));
second.ShouldBe(parent.CreateChild(1));
third.ShouldBe(parent.CreateChild(2));

// Flushless writes are visible in-memory immediately (no Flush required).
effect.Get<string>(first).ShouldBe("a");
effect.Get<string>(second).ShouldBe("b");
effect.Get<string>(third).ShouldBe("c");
}

[TestMethod]
public void CreateNextChildContinuesFromExistingChildren()
{
var parent = new EffectId([7]);
var existing = new List<StoredEffect>
{
StoredEffect.CreateCompleted(parent.CreateChild(0), alias: null),
StoredEffect.CreateCompleted(parent.CreateChild(1), alias: null),
};
var effect = CreateEffect(existing);

var next = effect.FlushlessCreateNextChild(parent, content: "x");

next.ShouldBe(parent.CreateChild(2));
}

[TestMethod]
public async Task CreateNextChildKeepsGapAndUsesHighestPlusOne()
{
var effect = CreateEffect(existingEffects: new List<StoredEffect>());
var parent = new EffectId([1]);

effect.FlushlessCreateNextChild(parent, content: "a"); // [1,0]
var middle = effect.FlushlessCreateNextChild(parent, content: "b"); // [1,1]
effect.FlushlessCreateNextChild(parent, content: "c"); // [1,2]

await effect.Clear(middle, flush: true); // remove [1,1], leaving {0,2}

var next = effect.FlushlessCreateNextChild(parent, content: "d");

// Highest existing index (2) + 1 -> gap at index 1 is kept.
next.ShouldBe(parent.CreateChild(3));
}

[TestMethod]
public void CreateNextChildIsScopedToItsParent()
{
var effect = CreateEffect(existingEffects: new List<StoredEffect>());
var parentA = new EffectId([1]);
var parentB = new EffectId([2]);

var a0 = effect.FlushlessCreateNextChild(parentA, content: "a0");
var a1 = effect.FlushlessCreateNextChild(parentA, content: "a1");
var b0 = effect.FlushlessCreateNextChild(parentB, content: "b0");

a0.ShouldBe(parentA.CreateChild(0));
a1.ShouldBe(parentA.CreateChild(1));
b0.ShouldBe(parentB.CreateChild(0));
}
}
9 changes: 9 additions & 0 deletions Core/Cleipnir.ResilientFunctions/Domain/Effect.cs
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,15 @@ internal T Get<T>(EffectId effectId)

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

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

#region Implicit ids

public Task Capture(Action work, ResiliencyLevel resiliency = ResiliencyLevel.AtLeastOnce)
Expand Down
27 changes: 26 additions & 1 deletion Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,26 @@ public IReadOnlyList<EffectId> GetChildren(EffectId parentId)
.Where(id => id.IsDescendant(parentId))
.ToList();
}


public EffectId FlushlessCreateNextChild<T>(EffectId parentId, T value, string? alias)
{
lock (_sync)
{
// Highest existing direct-child index + 1 (append). Index-selection and reservation
// happen under a single lock so concurrent callers cannot pick the same index.
var nextIndex = 0;
foreach (var id in _effectResults.Keys)
if (parentId.IsChild(id) && id.Id >= nextIndex)
nextIndex = id.Id + 1;

var childId = parentId.CreateChild(nextIndex);
var serializedValue = _serializer.Serialize(value!, typeof(T));
var storedEffect = StoredEffect.CreateCompleted(childId, serializedValue, alias);
AddToPendingCore(childId, storedEffect, delete: false, clearChildren: false);
return childId;
}
}

public async Task InnerCapture(EffectId effectId, string? alias, Func<Task> work, ResiliencyLevel resiliency, EffectContext effectContext)
{
EffectContext.SetParent(effectId);
Expand Down Expand Up @@ -417,6 +436,12 @@ private void AddToPending(IEnumerable<StoredEffect> storedEffects)
private void AddToPending(EffectId effectId, StoredEffect? storedEffect, bool delete, bool clearChildren)
{
lock (_sync)
AddToPendingCore(effectId, storedEffect, delete, clearChildren);
}

// Assumes _sync is already held by the caller.
private void AddToPendingCore(EffectId effectId, StoredEffect? storedEffect, bool delete, bool clearChildren)
{
{
if (_effectResults.ContainsKey(effectId))
{
Expand Down
Loading