Skip to content

Commit adca4a5

Browse files
dei79dei79
committed
Keep task table routing stable across month changes
1 parent f406b5b commit adca4a5

3 files changed

Lines changed: 158 additions & 25 deletions

File tree

‎dotNet/CoreHelpers.TaskLogging.Tests/AzureStorageTableTaskLoggerFactoryTests.cs‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,81 @@
11
using Azure;
2+
using Azure.Data.Tables;
23
using Xunit;
34

45
namespace CoreHelpers.TaskLogging.Tests;
56

67
public sealed class AzureStorageTableTaskLoggerFactoryTests
78
{
9+
[Theory]
10+
[InlineData("2026-01-31T23:59:59.9999999Z", 2026, 1)]
11+
[InlineData("2026-02-01T00:00:00Z", 2026, 2)]
12+
public void TaskKeyReferenceTime_PreservesAnnounceMonth(string timestamp, int expectedYear, int expectedMonth)
13+
{
14+
var taskKey = AzureTableTimebasedKeyBuilder.BuildDateTimeBasedRowKey(DateTimeOffset.Parse(timestamp), Guid.NewGuid().ToString());
15+
16+
var referenceTime = AzureTableTimebasedKeyBuilder.GetReferenceTime(taskKey);
17+
18+
Assert.Equal(expectedYear, referenceTime.Year);
19+
Assert.Equal(expectedMonth, referenceTime.Month);
20+
}
21+
22+
[Fact]
23+
public void TaskKeyReferenceTime_RejectsInvalidTaskKey()
24+
{
25+
var exception = Assert.Throws<ArgumentException>(() => AzureTableTimebasedKeyBuilder.GetReferenceTime("invalid-task-key"));
26+
27+
Assert.Equal("taskKey", exception.ParamName);
28+
}
29+
30+
[Fact]
31+
public void TimePartitionedTableNames_UseTaskKeyMonth()
32+
{
33+
var januaryTaskKey = AzureTableTimebasedKeyBuilder.BuildDateTimeBasedRowKey(new DateTimeOffset(2026, 1, 31, 23, 59, 59, TimeSpan.Zero), Guid.NewGuid().ToString());
34+
var februaryTaskKey = AzureTableTimebasedKeyBuilder.BuildDateTimeBasedRowKey(new DateTimeOffset(2026, 2, 1, 0, 0, 0, TimeSpan.Zero), Guid.NewGuid().ToString());
35+
36+
Assert.Equal("Dev202601Tasks", AzureStorageTableTaskLoggerFactory.GetTimePartitionedTableName("Dev", "Tasks", januaryTaskKey));
37+
Assert.Equal("Dev202601Messages", AzureStorageTableTaskLoggerFactory.GetTimePartitionedTableName("Dev", "Messages", januaryTaskKey));
38+
Assert.Equal("Dev202602Tasks", AzureStorageTableTaskLoggerFactory.GetTimePartitionedTableName("Dev", "Tasks", februaryTaskKey));
39+
Assert.Equal("Dev202602Messages", AzureStorageTableTaskLoggerFactory.GetTimePartitionedTableName("Dev", "Messages", februaryTaskKey));
40+
}
41+
42+
[Fact]
43+
public async Task TaskWorkflows_KeepUsingAnnounceMonth()
44+
{
45+
var taskKey = AzureTableTimebasedKeyBuilder.BuildDateTimeBasedRowKey(new DateTimeOffset(2026, 1, 31, 23, 59, 59, TimeSpan.Zero), Guid.NewGuid().ToString());
46+
var factory = new CapturingAzureStorageTableTaskLoggerFactory();
47+
48+
await factory.UpdateTaskStatus(taskKey, TaskStatus.Running);
49+
await factory.UpdateTaskWorker(taskKey, "worker");
50+
await factory.UpdateTaskStatus(taskKey, TaskStatus.Failed);
51+
await factory.MergePendingMessages(DateTimeOffset.UtcNow, taskKey, new[] { "message" });
52+
53+
Assert.Equal(
54+
new[]
55+
{
56+
("Update", "Dev202601Tasks"),
57+
("Add", "DevTasksRunning"),
58+
("Update", "Dev202601Tasks"),
59+
("Update", "Dev202601Tasks"),
60+
("Add", "Dev202601TasksFailed"),
61+
("Delete", "DevTasksRunning"),
62+
("Transaction", "Dev202601Messages")
63+
},
64+
factory.Operations);
65+
}
66+
67+
[Fact]
68+
public async Task AnnounceTask_UsesGeneratedTaskKeyMonth()
69+
{
70+
var factory = new CapturingAzureStorageTableTaskLoggerFactory();
71+
72+
var taskKey = await factory.AnnounceTask("type", "source", "worker");
73+
74+
Assert.Equal(
75+
new[] { ("Add", AzureStorageTableTaskLoggerFactory.GetTimePartitionedTableName("Dev", "Tasks", taskKey)) },
76+
factory.Operations);
77+
}
78+
879
[Fact]
980
public void GetNextMessageBatchSize_LimitsBatchToOneHundredEntities()
1081
{
@@ -92,4 +163,38 @@ public async Task ExecuteTableOperation_WhenRetryFails_RethrowsRetryException()
92163
Assert.Same(retryException, thrownException);
93164
Assert.Equal(2, operationCalls);
94165
}
166+
167+
private sealed class CapturingAzureStorageTableTaskLoggerFactory : AzureStorageTableTaskLoggerFactory
168+
{
169+
public CapturingAzureStorageTableTaskLoggerFactory()
170+
: base("UseDevelopmentStorage=true", "Dev", 100, TimeSpan.FromMinutes(5))
171+
{
172+
}
173+
174+
public List<(string Operation, string TableName)> Operations { get; } = new();
175+
176+
protected override Task AddEntityToTable<T>(string tableName, T entity, CancellationToken cancellationToken = default)
177+
{
178+
Operations.Add(("Add", tableName));
179+
return Task.CompletedTask;
180+
}
181+
182+
protected override Task UpdateEntityInTable<T>(string tableName, T entity, CancellationToken cancellationToken = default)
183+
{
184+
Operations.Add(("Update", tableName));
185+
return Task.CompletedTask;
186+
}
187+
188+
protected override Task DeleteEntityByKeys(string tableName, string pKey, string rowKey, CancellationToken cancellationToken = default)
189+
{
190+
Operations.Add(("Delete", tableName));
191+
return Task.CompletedTask;
192+
}
193+
194+
protected override Task SubmitTransactionToTable(string tableName, IReadOnlyList<TableTransactionAction> actions, CancellationToken cancellationToken)
195+
{
196+
Operations.Add(("Transaction", tableName));
197+
return Task.CompletedTask;
198+
}
199+
}
95200
}

‎dotNet/CoreHelpers.TaskLogging/AzureStorageTableTaskLoggerFactory.cs‎

Lines changed: 27 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ public async Task<string> AnnounceTask(string taskType, string taskSource, strin
6161
};
6262

6363
// get the table name
64-
var tableName = GetTaskTable();
64+
var tableName = GetTaskTable(taskKey);
6565

6666
// add the entity
6767
await AddEntityToTable<AzureTableTaskEntity>(tableName, taskEntity, cancellationToken);
@@ -102,7 +102,7 @@ public async Task UpdateTaskStatus(string taskKey, TaskStatus taskStatus, string
102102
taskEntity.TaskEndDate = DateTimeOffset.UtcNow;
103103

104104
// get the table name
105-
var tableName = GetTaskTable();
105+
var tableName = GetTaskTable(taskKey);
106106

107107
// update the entity
108108
await UpdateEntityInTable<AzureTableTaskEntity>(tableName, taskEntity);
@@ -115,7 +115,7 @@ public async Task UpdateTaskStatus(string taskKey, TaskStatus taskStatus, string
115115
else if (taskStatus == TaskStatus.Failed)
116116
{
117117
// store the task in the poisioned table
118-
await AddEntityToTable<AzureTableTaskEntity>(GetFailedTaskTable(), taskEntity);
118+
await AddEntityToTable<AzureTableTaskEntity>(GetFailedTaskTable(taskKey), taskEntity);
119119

120120
// remove the task from running table
121121
await DeleteEntityByKeys(GetRunningTaskTable(), taskKey, taskKey);
@@ -133,7 +133,7 @@ public async Task UpdateTaskWorker(string taskId, string taskWorker)
133133
};
134134

135135
// get the table name
136-
var tableName = GetTaskTable();
136+
var tableName = GetTaskTable(taskId);
137137

138138
// update the entity
139139
await UpdateEntityInTable<AzureTableTaskEntity>(tableName, taskEntity);
@@ -194,10 +194,7 @@ public async Task MergePendingMessages(DateTimeOffset flushTime, string taskKey,
194194
return;
195195

196196
// get the table name
197-
var tableName = GetTaskMessagesTable();
198-
199-
// get the table client
200-
var tableClient = _tableServiceClient.GetTableClient(tableName: tableName);
197+
var tableName = GetTaskMessagesTable(taskKey);
201198

202199
// build the table transaction
203200
var addEntitiesBatch = messages.Select(m => new TableTransactionAction(
@@ -211,7 +208,7 @@ public async Task MergePendingMessages(DateTimeOffset flushTime, string taskKey,
211208
})).ToArray();
212209

213210
// create the entry
214-
await ExecuteTableOperation(cancellationToken => tableClient.SubmitTransactionAsync(addEntitiesBatch, cancellationToken), cancellationToken => tableClient.CreateIfNotExistsAsync(cancellationToken), CancellationToken.None);
211+
await SubmitTransactionToTable(tableName, addEntitiesBatch, CancellationToken.None);
215212
}
216213

217214
internal static int GetNextMessageBatchSize(string taskKey, IReadOnlyList<string> messages)
@@ -248,20 +245,26 @@ private string GetTablePrefix()
248245
private string GetTableName(string tableName)
249246
=> $"{GetTablePrefix()}{tableName}";
250247

251-
private string GetTaskTable()
252-
=> GetTableName("Tasks");
248+
private string GetTableName(string tableName, string taskKey)
249+
=> GetTimePartitionedTableName(_environmentPrefix, tableName, taskKey);
250+
251+
internal static string GetTimePartitionedTableName(string environmentPrefix, string tableName, string taskKey)
252+
=> $"{environmentPrefix}{AzureTableTimebasedKeyBuilder.GetReferenceTime(taskKey):yyyyMM}{tableName}";
253+
254+
private string GetTaskTable(string taskKey)
255+
=> GetTableName("Tasks", taskKey);
253256

254257
private string GetExternalTaskIdLookupTable()
255258
=> GetTableName("TasksExternalIdLookup");
256259

257260
private string GetRunningTaskTable()
258261
=> $"{_environmentPrefix}TasksRunning";
259262

260-
private string GetFailedTaskTable()
261-
=> GetTableName("TasksFailed");
263+
private string GetFailedTaskTable(string taskKey)
264+
=> GetTableName("TasksFailed", taskKey);
262265

263-
private string GetTaskMessagesTable()
264-
=> GetTableName("Messages");
266+
private string GetTaskMessagesTable(string taskKey)
267+
=> GetTableName("Messages", taskKey);
265268

266269
private string BuildNextLogEntryTimestamp()
267270
{
@@ -278,15 +281,21 @@ private string BuildNextLogEntryTimestamp()
278281
return string.Format("{0}-{1}", Convert.ToInt64(diff.TotalSeconds), localEntryCounter.ToString("00000000"));
279282
}
280283

281-
private async Task AddEntityToTable<T>(string tableName, T entity, CancellationToken cancellationToken = default) where T : ITableEntity
284+
protected virtual async Task AddEntityToTable<T>(string tableName, T entity, CancellationToken cancellationToken = default) where T : ITableEntity
282285
=> await ExecuteEntityToTableOperation(tableName, (TableClient tc, CancellationToken token) => tc.AddEntityAsync<T>(entity, token), cancellationToken);
283286

284-
private async Task UpdateEntityInTable<T>(string tableName, T entity, CancellationToken cancellationToken = default) where T : ITableEntity
287+
protected virtual async Task UpdateEntityInTable<T>(string tableName, T entity, CancellationToken cancellationToken = default) where T : ITableEntity
285288
=> await ExecuteEntityToTableOperation(tableName, (TableClient tc, CancellationToken token) => tc.UpdateEntityAsync<T>(entity, Azure.ETag.All, cancellationToken: token), cancellationToken);
286289

287-
private async Task DeleteEntityByKeys(string tableName, string pKey, string rowKey, CancellationToken cancellationToken = default)
290+
protected virtual async Task DeleteEntityByKeys(string tableName, string pKey, string rowKey, CancellationToken cancellationToken = default)
288291
=> await ExecuteEntityToTableOperation(tableName, (TableClient tc, CancellationToken token) => tc.DeleteEntityAsync(pKey, rowKey, Azure.ETag.All, token), cancellationToken);
289292

293+
protected virtual async Task SubmitTransactionToTable(string tableName, IReadOnlyList<TableTransactionAction> actions, CancellationToken cancellationToken)
294+
{
295+
var tableClient = _tableServiceClient.GetTableClient(tableName: tableName);
296+
await ExecuteTableOperation(token => tableClient.SubmitTransactionAsync(actions, token), token => tableClient.CreateIfNotExistsAsync(token), cancellationToken);
297+
}
298+
290299
private async Task<T[]> QueryEntitiyFromTableByPartitionKey<T>(string tableName, string partitionKey, CancellationToken cancellationToken = default) where T : class, ITableEntity
291300
{
292301
var result = new List<T>();

‎dotNet/CoreHelpers.TaskLogging/AzureTableTimebasedKeyBuilder.cs‎

Lines changed: 26 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,32 @@ namespace CoreHelpers.TaskLogging
55
{
66
public static class AzureTableTimebasedKeyBuilder
77
{
8+
private const long MaxMoment = 9007199254740991; // Number.MAX_SAFE_INTEGER
9+
private const string TaskKeyPrefix = "task";
10+
private const int EncodedMomentLength = 16;
11+
812
public static string BuildDateTimeBasedRowKey(DateTimeOffset refTime, string postfix)
913
{
10-
long maxMoment = 9007199254740991; // Number.MAX_SAFE_INTEGER
11-
DateTime origin = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
12-
TimeSpan diff = refTime.ToUniversalTime() - origin;
13-
return $"task{Convert.ToInt64(maxMoment - diff.TotalSeconds)}{postfix}";
14-
}
15-
}
16-
}
14+
return $"{TaskKeyPrefix}{MaxMoment - refTime.ToUnixTimeSeconds()}{postfix}";
15+
}
16+
17+
internal static DateTimeOffset GetReferenceTime(string taskKey)
18+
{
19+
if (string.IsNullOrEmpty(taskKey) || !taskKey.StartsWith(TaskKeyPrefix, StringComparison.Ordinal) || taskKey.Length < TaskKeyPrefix.Length + EncodedMomentLength)
20+
throw new ArgumentException("The task key does not contain a valid reference time.", nameof(taskKey));
21+
22+
var encodedMomentText = taskKey.Substring(TaskKeyPrefix.Length, EncodedMomentLength);
23+
if (!long.TryParse(encodedMomentText, out var encodedMoment))
24+
throw new ArgumentException("The task key does not contain a valid reference time.", nameof(taskKey));
1725

26+
try
27+
{
28+
return DateTimeOffset.FromUnixTimeSeconds(MaxMoment - encodedMoment);
29+
}
30+
catch (ArgumentOutOfRangeException exception)
31+
{
32+
throw new ArgumentException("The task key does not contain a valid reference time.", nameof(taskKey), exception);
33+
}
34+
}
35+
}
36+
}

0 commit comments

Comments
 (0)