Skip to content

Commit 7f183c1

Browse files
zhhyuzhhyu
authored
[ST] 3. Catalog2Registration: Read cursor updates (#10708)
* add * update * update * update * update
1 parent 6ddfec4 commit 7f183c1

5 files changed

Lines changed: 173 additions & 10 deletions

File tree

‎src/Catalog/HttpReadCursor.cs‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright (c) .NET Foundation. All rights reserved.
1+
// Copyright (c) .NET Foundation. All rights reserved.
22
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
33

44
using System;
@@ -14,9 +14,9 @@ namespace NuGet.Services.Metadata.Catalog
1414
{
1515
public class HttpReadCursor : ReadCursor
1616
{
17-
private readonly Uri _address;
18-
private readonly DateTime? _defaultValue;
19-
private readonly Func<HttpMessageHandler> _handlerFunc;
17+
protected readonly Uri _address;
18+
protected readonly DateTime? _defaultValue;
19+
protected readonly Func<HttpMessageHandler> _handlerFunc;
2020

2121
public HttpReadCursor(Uri address, DateTime defaultValue, Func<HttpMessageHandler> handlerFunc = null)
2222
{
@@ -52,9 +52,9 @@ await Retry.IncrementalAsync(
5252
{
5353
response.EnsureSuccessStatusCode();
5454

55-
string json = await response.Content.ReadAsStringAsync();
55+
string valueInJson = await GetValueInJsonAsync(response);
5656

57-
JObject obj = JObject.Parse(json);
57+
JObject obj = JObject.Parse(valueInJson);
5858
Value = obj["value"].ToObject<DateTime>();
5959
}
6060
}
@@ -66,5 +66,10 @@ await Retry.IncrementalAsync(
6666
initialWaitInterval: TimeSpan.Zero,
6767
waitIncrement: TimeSpan.FromSeconds(10));
6868
}
69+
70+
public virtual async Task<string> GetValueInJsonAsync(HttpResponseMessage response)
71+
{
72+
return await response.Content.ReadAsStringAsync();
73+
}
6974
}
70-
}
75+
}

‎src/Catalog/HttpReadCursorWithUpdates.cs‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
// Copyright (c) .NET Foundation. All rights reserved.
2+
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
3+
4+
using System;
5+
using System.Linq;
6+
using System.Net.Http;
7+
using System.Threading.Tasks;
8+
using Microsoft.Extensions.Logging;
9+
using Newtonsoft.Json;
10+
11+
namespace NuGet.Services.Metadata.Catalog
12+
{
13+
public class HttpReadCursorWithUpdates : HttpReadCursor
14+
{
15+
private readonly ILogger _logger;
16+
private readonly TimeSpan _minIntervalBeforeToReadCursorUpdateValue;
17+
18+
public HttpReadCursorWithUpdates(TimeSpan minIntervalBeforeToReadCursorUpdateValue, Uri address, ILogger logger,
19+
Func<HttpMessageHandler> handlerFunc = null)
20+
: base(address, handlerFunc)
21+
{
22+
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
23+
_minIntervalBeforeToReadCursorUpdateValue = minIntervalBeforeToReadCursorUpdateValue;
24+
}
25+
26+
public override async Task<string> GetValueInJsonAsync(HttpResponseMessage response)
27+
{
28+
var storageDateTimeOffset = response.Headers.Date;
29+
DateTime? storageDateTimeInUtc = null;
30+
if (storageDateTimeOffset.HasValue)
31+
{
32+
storageDateTimeInUtc = storageDateTimeOffset.Value.UtcDateTime;
33+
}
34+
35+
var update = await GetUpdate(response, storageDateTimeInUtc);
36+
37+
return JsonConvert.SerializeObject(new { value = update.Value });
38+
}
39+
40+
private async Task<CursorValueUpdate> GetUpdate(HttpResponseMessage response, DateTime? storageDateTimeInUtc)
41+
{
42+
if (!storageDateTimeInUtc.HasValue)
43+
{
44+
throw new ArgumentNullException(nameof(storageDateTimeInUtc));
45+
}
46+
47+
var content = await response.Content.ReadAsStringAsync();
48+
var cursorValueWithUpdates = JsonConvert.DeserializeObject<CursorValueWithUpdates>(content, CursorValueWithUpdates.SerializerSettings);
49+
var updates = cursorValueWithUpdates.Updates.OrderByDescending(u => u.TimeStamp).ToList();
50+
51+
foreach (var update in updates)
52+
{
53+
if (update.TimeStamp <= storageDateTimeInUtc.Value - _minIntervalBeforeToReadCursorUpdateValue)
54+
{
55+
_logger.LogInformation("Read the cursor update with timeStamp: {TimeStamp} and value: {UpdateValue}, at {Address}. " +
56+
"(Storage DateTime: {StorageDateTime}, MinIntervalBeforeToReadCursorUpdateValue: {MinIntervalBeforeToReadCursorUpdateValue})",
57+
update.TimeStamp.ToString(CursorValueWithUpdates.SerializerSettings.DateFormatString),
58+
update.Value,
59+
_address.AbsoluteUri,
60+
storageDateTimeInUtc.Value.ToString(CursorValueWithUpdates.SerializerSettings.DateFormatString),
61+
_minIntervalBeforeToReadCursorUpdateValue);
62+
63+
return update;
64+
}
65+
}
66+
67+
if (updates.Count > 0)
68+
{
69+
_logger.LogWarning("Unable to find the cursor update and the oldest cursor update has timeStamp: {TimeStamp}, at {Address}. " +
70+
"(Storage DateTime: {StorageDateTime}, MinIntervalBeforeToReadCursorUpdateValue: {MinIntervalBeforeToReadCursorUpdateValue})",
71+
updates.Last().TimeStamp.ToString(CursorValueWithUpdates.SerializerSettings.DateFormatString),
72+
_address.AbsoluteUri,
73+
storageDateTimeInUtc.Value.ToString(CursorValueWithUpdates.SerializerSettings.DateFormatString),
74+
_minIntervalBeforeToReadCursorUpdateValue);
75+
}
76+
else
77+
{
78+
_logger.LogWarning("Unable to find the cursor update and the count of updates is {CursorUpdatesCount}, at {Address}.",
79+
updates.Count,
80+
_address.AbsoluteUri);
81+
}
82+
83+
throw new InvalidOperationException("Unable to find the cursor update.");
84+
}
85+
}
86+
}

‎src/NuGet.Jobs.Catalog2Registration/Catalog2RegistrationCommand.cs‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright (c) .NET Foundation. All rights reserved.
1+
// Copyright (c) .NET Foundation. All rights reserved.
22
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
33

44
using System;
@@ -10,6 +10,7 @@
1010
using Microsoft.Extensions.Options;
1111
using Microsoft.WindowsAzure.Storage;
1212
using NuGet.Services.Metadata.Catalog;
13+
using NuGet.Services.Metadata.Catalog.Dnx;
1314
using NuGet.Services.Metadata.Catalog.Persistence;
1415
using NuGet.Services.V3;
1516
using NuGetGallery;
@@ -58,7 +59,9 @@ private async Task ExecuteAsync(CancellationToken token)
5859
_logger.LogInformation("Depending on cursors: {DependencyCursorUrls}", _options.Value.DependencyCursorUrls);
5960
backCursor = new AggregateCursor(_options
6061
.Value
61-
.DependencyCursorUrls.Select(r => new HttpReadCursor(new Uri(r), _handlerFunc)));
62+
.DependencyCursorUrls.Select(r => new HttpReadCursorWithUpdates(
63+
minIntervalBeforeToReadCursorUpdateValue: DnxConstants.CacheDurationOfPackageVersionIndex + TimeSpan.FromSeconds(1),
64+
new Uri(r), _logger, _handlerFunc)));
6265
}
6366
else
6467
{

‎tests/CatalogTests/HttpReadCursorWithUpdatesTests.cs‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
// Copyright (c) .NET Foundation. All rights reserved.
2+
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
3+
4+
using System;
5+
using System.Net.Http;
6+
using System.Threading.Tasks;
7+
using Microsoft.Extensions.Logging;
8+
using Moq;
9+
using NuGet.Services.Metadata.Catalog;
10+
using Xunit;
11+
12+
namespace CatalogTests
13+
{
14+
public class HttpReadCursorWithUpdatesTests
15+
{
16+
private readonly HttpReadCursorWithUpdates _cursor;
17+
private readonly HttpResponseMessage _response;
18+
19+
public HttpReadCursorWithUpdatesTests()
20+
{
21+
_cursor = new HttpReadCursorWithUpdates(minIntervalBeforeToReadCursorUpdateValue: TimeSpan.FromMinutes(1) + TimeSpan.FromSeconds(1),
22+
new Uri("https://test"), Mock.Of<ILogger>());
23+
_response = new HttpResponseMessage();
24+
}
25+
26+
[Fact]
27+
public async Task GetValueInJsonAsync()
28+
{
29+
_response.Headers.Date = new DateTimeOffset(2026, 1, 1, 1, 0, 30, TimeSpan.Zero);
30+
_response.Content = new StringContent("{\"value\":\"2026-01-01T01:00:00.0000000\"," +
31+
"\"updates\":[{\"timeStamp\":\"2026-01-01T00:59:30.0000000Z\",\"value\":\"2026-01-01T00:59:00.0000000\"}," +
32+
"{\"timeStamp\":\"2026-01-01T00:58:30.0000000Z\",\"value\":\"2026-01-01T00:58:00.0000000\"}," +
33+
"{\"timeStamp\":\"2026-01-01T00:57:30.0000000Z\",\"value\":\"2026-01-01T00:57:00.0000000\"}]}");
34+
35+
Assert.Equal("{\"value\":\"2026-01-01T00:58:00.0000000\"}", await _cursor.GetValueInJsonAsync(_response));
36+
}
37+
38+
[Theory]
39+
[InlineData("{\"value\":\"2026-01-01T01:00:00.0000000\"}")]
40+
[InlineData("{\"value\":\"2026-01-01T01:00:00.0000000\",\"updates\":[]}")]
41+
[InlineData("{\"value\":\"2026-01-01T01:00:00.0000000\"," +
42+
"\"updates\":[{\"timeStamp\":\"2026-01-01T00:59:30.0000000Z\",\"value\":\"2026-01-01T00:59:00.0000000\"}]}")]
43+
public async Task GetValueInJsonAsync_UnableToFindCursorUpdate(string content)
44+
{
45+
_response.Headers.Date = new DateTimeOffset(2026, 1, 1, 1, 0, 30, TimeSpan.Zero);
46+
_response.Content = new StringContent(content);
47+
48+
var exception = await Assert.ThrowsAsync<InvalidOperationException>(() => _cursor.GetValueInJsonAsync(_response));
49+
Assert.Equal("Unable to find the cursor update.", exception.Message);
50+
}
51+
52+
[Fact]
53+
public async Task GetValueInJsonAsync_WithoutStorageDateTime()
54+
{
55+
_response.Headers.Date = null;
56+
57+
var exception = await Assert.ThrowsAsync<ArgumentNullException>(() => _cursor.GetValueInJsonAsync(_response));
58+
Assert.Equal("storageDateTimeInUtc", exception.ParamName);
59+
}
60+
}
61+
}

‎tests/NuGet.Jobs.Catalog2Registration.Tests/Catalog2RegistrationCommandFacts.cs‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
using Moq.Protected;
1313
using NuGet.Services;
1414
using NuGet.Services.Metadata.Catalog;
15+
using NuGet.Services.Metadata.Catalog.Dnx;
1516
using NuGet.Services.Metadata.Catalog.Persistence;
1617
using NuGet.Services.V3;
1718
using NuGetGallery;
@@ -74,11 +75,18 @@ public async Task LoadsDependencyCursorsIfConfigured()
7475
"https://example/fc-1/cursor.json",
7576
"https://example/fc-2/cursor.json",
7677
};
78+
79+
var responseDate = new DateTimeOffset(2026, 1, 1, 1, 0, 30, TimeSpan.Zero);
80+
var updateTimeStamp = (responseDate.UtcDateTime - (DnxConstants.CacheDurationOfPackageVersionIndex + TimeSpan.FromSeconds(1)))
81+
.ToString(CursorValueWithUpdates.SerializerSettings.DateFormatString);
82+
7783
HttpMessageHandler
7884
.Setup(x => x.OnSendAsync(It.IsAny<HttpRequestMessage>(), It.IsAny<CancellationToken>()))
7985
.ReturnsAsync(() => new HttpResponseMessage(HttpStatusCode.OK)
8086
{
81-
Content = new StringContent($"{{\"value\": \"{DateTimeOffset.UtcNow.ToString("O")}\"}}"),
87+
Headers = { Date = responseDate },
88+
Content = new StringContent($"{{\"value\": \"2026-01-01T00:00:00.0000000\"," +
89+
$"\"updates\":[{{\"timeStamp\":\"{ updateTimeStamp }\",\"value\":\"2026-01-01T00:00:00.0000000\"}}]}}"),
8290
});
8391

8492
await Target.ExecuteAsync();

0 commit comments

Comments
 (0)