Skip to content

Commit a3cfea5

Browse files
eamsdeneamsden
committed
Lots of fixes
1 parent a2e8a73 commit a3cfea5

12 files changed

Lines changed: 511 additions & 40 deletions

File tree

‎.github/workflows/package.yml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,10 @@ jobs:
170170
src/Temporalio/bin/Release/*.snupkg
171171
src/Temporalio.Extensions.DiagnosticSource/bin/Release/*.nupkg
172172
src/Temporalio.Extensions.DiagnosticSource/bin/Release/*.snupkg
173+
src/Temporalio.Extensions.Aws.Lambda/bin/Release/*.nupkg
174+
src/Temporalio.Extensions.Aws.Lambda/bin/Release/*.snupkg
175+
src/Temporalio.Extensions.Aws.Lambda.OpenTelemetry/bin/Release/*.nupkg
176+
src/Temporalio.Extensions.Aws.Lambda.OpenTelemetry/bin/Release/*.snupkg
173177
src/Temporalio.Extensions.Hosting/bin/Release/*.nupkg
174178
src/Temporalio.Extensions.Hosting/bin/Release/*.snupkg
175179
src/Temporalio.Extensions.OpenTelemetry/bin/Release/*.nupkg

‎src/Temporalio.ApiDoc/docfx.json‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@
66
"files": [
77
"Temporalio/*.csproj",
88
"Temporalio.Extensions.DiagnosticSource/*.csproj",
9+
"Temporalio.Extensions.Aws.Lambda/*.csproj",
10+
"Temporalio.Extensions.Aws.Lambda.OpenTelemetry/*.csproj",
911
"Temporalio.Extensions.Hosting/*.csproj",
1012
"Temporalio.Extensions.OpenTelemetry/*.csproj"
1113
],
@@ -72,4 +74,4 @@
7274
"keepFileLink": false,
7375
"disableGitFeatures": false
7476
}
75-
}
77+
}

‎src/Temporalio.Extensions.Aws.Lambda.OpenTelemetry/LambdaWorkerOpenTelemetry.cs‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -110,14 +110,30 @@ internal static async Task ForceFlushAsync(
110110
return;
111111
}
112112

113-
await Task.Run(
114-
() => tracerProvider.ForceFlush(ToTimeoutMilliseconds(shutdownDeadlineBuffer))).
115-
ConfigureAwait(false);
113+
var flushTask = Task.Run(
114+
() => tracerProvider.ForceFlush(ToTimeoutMilliseconds(shutdownDeadlineBuffer)));
115+
if (flushTask == await Task.WhenAny(
116+
flushTask,
117+
Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false))
118+
{
119+
await flushTask.ConfigureAwait(false);
120+
}
121+
else
122+
{
123+
ObserveTaskException(flushTask);
124+
}
116125
}
117126

118127
private static string FirstNonEmpty(params string?[] values) =>
119128
values.First(value => !string.IsNullOrEmpty(value))!;
120129

130+
private static void ObserveTaskException(Task task) =>
131+
_ = task.ContinueWith(
132+
completedTask => _ = completedTask.Exception,
133+
CancellationToken.None,
134+
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
135+
TaskScheduler.Default);
136+
121137
private static TracerProvider CreateTracerProvider(
122138
ResolvedLambdaWorkerOpenTelemetryOptions options) =>
123139
Sdk.CreateTracerProviderBuilder().

‎src/Temporalio.Extensions.Aws.Lambda/LambdaWorkerConfig.cs‎

Lines changed: 39 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@
44
using System.Threading.Tasks;
55
using Temporalio.Client;
66
using Temporalio.Worker;
7-
using Temporalio.Worker.Tuning;
87

98
namespace Temporalio.Extensions.Aws.Lambda
109
{
@@ -18,27 +17,62 @@ public class LambdaWorkerConfig
1817
/// </summary>
1918
public static readonly TimeSpan DefaultShutdownDeadlineBuffer = TimeSpan.FromSeconds(7);
2019

20+
private Func<TemporalClientConnectOptions>? loadClientOptions;
21+
private TemporalClientConnectOptions? clientOptions;
22+
private bool clientOptionsSet;
23+
2124
/// <summary>
2225
/// Initializes a new instance of the <see cref="LambdaWorkerConfig"/> class.
2326
/// </summary>
2427
public LambdaWorkerConfig()
28+
: this(null)
29+
{
30+
}
31+
32+
/// <summary>
33+
/// Initializes a new instance of the <see cref="LambdaWorkerConfig"/> class.
34+
/// </summary>
35+
/// <param name="loadClientOptions">Lazy client options loader.</param>
36+
internal LambdaWorkerConfig(Func<TemporalClientConnectOptions>? loadClientOptions)
2537
{
38+
this.loadClientOptions = loadClientOptions;
2639
WorkerOptions.MaxConcurrentActivities = 2;
2740
WorkerOptions.MaxConcurrentWorkflowTasks = 10;
2841
WorkerOptions.MaxConcurrentLocalActivities = 2;
2942
WorkerOptions.MaxConcurrentNexusTasks = 5;
3043
WorkerOptions.GracefulShutdownTimeout = TimeSpan.FromSeconds(5);
3144
WorkerOptions.MaxCachedWorkflows = 30;
32-
WorkerOptions.WorkflowTaskPollerBehavior = new PollerBehavior.SimpleMaximum(2);
33-
WorkerOptions.ActivityTaskPollerBehavior = new PollerBehavior.SimpleMaximum(1);
34-
WorkerOptions.NexusTaskPollerBehavior = new PollerBehavior.SimpleMaximum(1);
45+
WorkerOptions.MaxConcurrentWorkflowTaskPolls = 2;
46+
WorkerOptions.MaxConcurrentActivityTaskPolls = 1;
47+
WorkerOptions.MaxConcurrentNexusTaskPolls = 1;
3548
WorkerOptions.DisableEagerActivityExecution = true;
3649
}
3750

3851
/// <summary>
3952
/// Gets or sets the client connection options.
4053
/// </summary>
41-
public TemporalClientConnectOptions ClientOptions { get; set; } = new TemporalClientConnectOptions();
54+
public TemporalClientConnectOptions ClientOptions
55+
{
56+
get
57+
{
58+
if (!clientOptionsSet)
59+
{
60+
clientOptions = loadClientOptions == null ?
61+
new TemporalClientConnectOptions() :
62+
loadClientOptions();
63+
loadClientOptions = null;
64+
clientOptionsSet = true;
65+
}
66+
return clientOptions!;
67+
}
68+
69+
set
70+
{
71+
clientOptions = value;
72+
loadClientOptions = null;
73+
clientOptionsSet = true;
74+
}
75+
}
4276

4377
/// <summary>
4478
/// Gets or sets the worker options.

‎src/Temporalio.Extensions.Aws.Lambda/README.md‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -74,9 +74,9 @@ queue in `configure`. If no task queue is set by the environment or by `configur
7474

7575
`TemporalLambdaWorker.CreateHandler` requires a `WorkerDeploymentVersion` and always enables Worker Versioning by setting
7676
`WorkerOptions.DeploymentOptions` with `UseWorkerVersioning = true`. Use a deployment name and build ID that match your
77-
rollout process. If you need to set the default versioning behavior, configure
78-
`config.WorkerOptions.DeploymentOptions.DefaultVersioningBehavior`; the handler preserves that value while enforcing the
79-
deployment version passed to `CreateHandler`.
77+
rollout process. The default versioning behavior is `AutoUpgrade`. If you need a different default versioning behavior,
78+
configure `config.WorkerOptions.DeploymentOptions.DefaultVersioningBehavior`; the handler preserves any non-`Unspecified`
79+
value while enforcing the deployment version passed to `CreateHandler`.
8080

8181
The helper applies Lambda-oriented worker defaults before `configure`, including lower concurrency, a 5 second graceful
8282
shutdown timeout, a smaller workflow cache, simple poller limits, and disabled eager activity execution. Values you set in

‎src/Temporalio.Extensions.Aws.Lambda/TemporalLambdaWorker.cs‎

Lines changed: 44 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -102,18 +102,17 @@ public static TemporalClientConnectOptions LoadClientConnectOptions(
102102
throw new ArgumentException("Build ID must be set", nameof(version));
103103
}
104104

105-
var config = new LambdaWorkerConfig
106-
{
107-
ClientOptions = handlerOptions.LoadClientConnectOptions == null ?
108-
new TemporalClientConnectOptions() :
109-
handlerOptions.LoadClientConnectOptions(null),
110-
};
105+
var loadClientConnectOptions = handlerOptions.LoadClientConnectOptions;
106+
var config = new LambdaWorkerConfig(
107+
loadClientConnectOptions == null ?
108+
null :
109+
() => loadClientConnectOptions(null));
111110
var environmentTaskQueue = handlerOptions.GetEnvironmentVariable(TaskQueueEnvironmentVariable);
112111
if (environmentTaskQueue != null)
113112
{
114113
config.WorkerOptions.TaskQueue = environmentTaskQueue;
115114
}
116-
config.WorkerOptions.DeploymentOptions = new WorkerDeploymentOptions(version, useWorkerVersioning: true);
115+
ApplyDeploymentVersion(config.WorkerOptions, version);
117116

118117
configure(config);
119118
var state = PrepareHandlerState(version, config, handlerOptions);
@@ -147,17 +146,14 @@ private static LambdaWorkerHandlerState PrepareHandlerState(
147146
"WorkerOptions.TaskQueue must be set or TEMPORAL_TASK_QUEUE must be present");
148147
}
149148

150-
var defaultVersioningBehavior =
151-
config.WorkerOptions.DeploymentOptions?.DefaultVersioningBehavior ??
152-
VersioningBehavior.Unspecified;
153-
config.WorkerOptions.DeploymentOptions = new WorkerDeploymentOptions(
154-
version,
155-
useWorkerVersioning: true)
149+
var postPluginConfiguration = config.WorkerOptions.PostPluginConfiguration;
150+
config.WorkerOptions.PostPluginConfiguration = options =>
156151
{
157-
DefaultVersioningBehavior = defaultVersioningBehavior,
152+
postPluginConfiguration?.Invoke(options);
153+
ApplyDeploymentVersion(options, version);
154+
ClearConcurrencyLimitsIfTunerSet(options);
158155
};
159-
config.WorkerOptions.BuildId = null;
160-
config.WorkerOptions.UseWorkerVersioning = false;
156+
config.WorkerOptions.ApplyPostPluginConfiguration();
161157

162158
foreach (var hook in config.ShutdownHooks)
163159
{
@@ -175,6 +171,38 @@ private static LambdaWorkerHandlerState PrepareHandlerState(
175171
handlerOptions);
176172
}
177173

174+
private static void ApplyDeploymentVersion(
175+
TemporalWorkerOptions workerOptions,
176+
WorkerDeploymentVersion version)
177+
{
178+
var defaultVersioningBehavior =
179+
workerOptions.DeploymentOptions?.DefaultVersioningBehavior is { } behavior &&
180+
behavior != VersioningBehavior.Unspecified ?
181+
behavior :
182+
VersioningBehavior.AutoUpgrade;
183+
workerOptions.DeploymentOptions = new WorkerDeploymentOptions(
184+
version,
185+
useWorkerVersioning: true)
186+
{
187+
DefaultVersioningBehavior = defaultVersioningBehavior,
188+
};
189+
workerOptions.BuildId = null;
190+
workerOptions.UseWorkerVersioning = false;
191+
}
192+
193+
private static void ClearConcurrencyLimitsIfTunerSet(TemporalWorkerOptions workerOptions)
194+
{
195+
if (workerOptions.Tuner == null)
196+
{
197+
return;
198+
}
199+
200+
workerOptions.MaxConcurrentActivities = null;
201+
workerOptions.MaxConcurrentWorkflowTasks = null;
202+
workerOptions.MaxConcurrentLocalActivities = null;
203+
workerOptions.MaxConcurrentNexusTasks = null;
204+
}
205+
178206
private static CancellationTokenSource CreateHookCancellationTokenSource(
179207
TimeSpan remainingTime)
180208
{

‎src/Temporalio/Temporalio.csproj‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,9 @@
2828
<AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
2929
<_Parameter1>Temporalio.Tests</_Parameter1>
3030
</AssemblyAttribute>
31+
<AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
32+
<_Parameter1>Temporalio.Extensions.Aws.Lambda</_Parameter1>
33+
</AssemblyAttribute>
3134
</ItemGroup>
3235

3336
<!-- Build/copy debug/release bridge DLL -->

‎src/Temporalio/Worker/TemporalWorker.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ public TemporalWorker(IWorkerClient client, TemporalWorkerOptions options)
5757
{
5858
plugin.ConfigureWorker(Options);
5959
}
60+
Options.ApplyPostPluginConfiguration();
6061

6162
// Ensure later accesses use the modified version of options.
6263
options = Options;

‎src/Temporalio/Worker/TemporalWorkerOptions.cs‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -376,6 +376,14 @@ public TemporalWorkerOptions()
376376
internal Func<WorkflowInstanceDetails, IWorkflowInstance> WorkflowInstanceFactory { get; set; } =
377377
DefaultWorkflowInstanceFactory;
378378

379+
/// <summary>
380+
/// Gets or sets a function to run after worker plugins configure options.
381+
/// </summary>
382+
/// <remarks>
383+
/// Don't expose this until there's a use case.
384+
/// </remarks>
385+
internal Action<TemporalWorkerOptions>? PostPluginConfiguration { get; set; }
386+
379387
/// <summary>
380388
/// Add the given delegate with <see cref="ActivityAttribute" /> as an activity. This is
381389
/// usually a method reference.
@@ -517,5 +525,10 @@ internal void OnTaskCompleted(WorkflowInstance instance, Exception? failureExcep
517525
handler(instance, new(instance, failureException));
518526
}
519527
}
528+
529+
/// <summary>
530+
/// Run post-plugin configuration.
531+
/// </summary>
532+
internal void ApplyPostPluginConfiguration() => PostPluginConfiguration?.Invoke(this);
520533
}
521534
}

‎tests/Temporalio.Tests/Extensions/Aws/Lambda/OpenTelemetry/LambdaWorkerOpenTelemetryTests.cs‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -194,18 +194,56 @@ public async Task ForceFlushAsync_RunsForceFlushOffCallerThread()
194194
}
195195
}
196196

197+
[Fact]
198+
public async Task ForceFlushAsync_ReturnsWhenCancellationRequested()
199+
{
200+
using var flushStarted = new ManualResetEventSlim();
201+
using var releaseFlush = new ManualResetEventSlim();
202+
using var flushCompleted = new ManualResetEventSlim();
203+
#pragma warning disable CA2000 // Tracer provider owns the processor/exporter.
204+
using var provider = Sdk.CreateTracerProviderBuilder().
205+
AddProcessor(new SimpleActivityExportProcessor(
206+
new BlockingForceFlushExporter(flushStarted, releaseFlush, flushCompleted))).
207+
Build();
208+
#pragma warning restore CA2000
209+
using var cts = new CancellationTokenSource();
210+
211+
#pragma warning disable CA2025 // The provider exits scope after the blocking flush is released.
212+
var flushTask = LambdaWorkerOpenTelemetry.ForceFlushAsync(
213+
provider,
214+
TimeSpan.FromSeconds(10),
215+
cts.Token);
216+
#pragma warning restore CA2025
217+
218+
try
219+
{
220+
Assert.True(flushStarted.Wait(TimeSpan.FromSeconds(5)));
221+
await cts.CancelAsync();
222+
await flushTask.WaitAsync(TimeSpan.FromSeconds(5));
223+
Assert.False(flushCompleted.IsSet);
224+
}
225+
finally
226+
{
227+
releaseFlush.Set();
228+
Assert.True(flushCompleted.Wait(TimeSpan.FromSeconds(5)));
229+
}
230+
}
231+
197232
private sealed class BlockingForceFlushExporter :
198233
BaseExporter<System.Diagnostics.Activity>
199234
{
200235
private readonly ManualResetEventSlim flushStarted;
201236
private readonly ManualResetEventSlim releaseFlush;
237+
private readonly ManualResetEventSlim? flushCompleted;
202238

203239
public BlockingForceFlushExporter(
204240
ManualResetEventSlim flushStarted,
205-
ManualResetEventSlim releaseFlush)
241+
ManualResetEventSlim releaseFlush,
242+
ManualResetEventSlim? flushCompleted = null)
206243
{
207244
this.flushStarted = flushStarted;
208245
this.releaseFlush = releaseFlush;
246+
this.flushCompleted = flushCompleted;
209247
}
210248

211249
public override ExportResult Export(in Batch<System.Diagnostics.Activity> batch) =>
@@ -214,7 +252,14 @@ public override ExportResult Export(in Batch<System.Diagnostics.Activity> batch)
214252
protected override bool OnForceFlush(int timeoutMilliseconds)
215253
{
216254
flushStarted.Set();
217-
return releaseFlush.Wait(timeoutMilliseconds);
255+
try
256+
{
257+
return releaseFlush.Wait(timeoutMilliseconds);
258+
}
259+
finally
260+
{
261+
flushCompleted?.Set();
262+
}
218263
}
219264
}
220265

0 commit comments

Comments
 (0)