refactor: replace per-client PIT with shared singleton PIT manager

Extract PIT lifecycle from ChangesGateway into SharedPointInTimeManager,
a thread-safe singleton that holds a single PIT for all concurrent
paginators. Reduces ES resource usage for the public changes API.
The cursor no longer carries a PIT ID — it's just [epochMs, url].

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: reakaleek

src/api/Elastic.Documentation.Api.Core/Changes/ChangesUsecase.cs

@@ -69,11 +69,7 @@ public async Task<ChangesApiResponse> GetChangesAsync(ChangesApiRequest request,
 			if (epochEl.ValueKind != JsonValueKind.Number || urlEl.ValueKind != JsonValueKind.String)
 				return null;
 
-			var pitId = arrayLength > 2 && root[2].ValueKind == JsonValueKind.String
-				? root[2].GetString()
-				: null;
-
-			return new ChangesPageCursor(epochEl.GetInt64(), urlEl.GetString()!, pitId);
+			return new ChangesPageCursor(epochEl.GetInt64(), urlEl.GetString()!);
 		}
 		catch (Exception ex) when (ex is FormatException or JsonException or InvalidOperationException)
 		{
@@ -88,8 +84,6 @@ private static string EncodeCursor(ChangesPageCursor cursor)
 		writer.WriteStartArray();
 		writer.WriteNumberValue(cursor.LastUpdatedEpochMs);
 		writer.WriteStringValue(cursor.Url);
-		if (cursor.PitId is not null)
-			writer.WriteStringValue(cursor.PitId);
 		writer.WriteEndArray();
 		writer.Flush();
 

src/api/Elastic.Documentation.Api.Core/Changes/IChangesGateway.cs

@@ -26,7 +26,7 @@ public record ChangesResult
 }
 
 /// <summary>Cursor for search_after pagination over changed pages.</summary>
-public record ChangesPageCursor(long LastUpdatedEpochMs, string Url, string? PitId = null);
+public record ChangesPageCursor(long LastUpdatedEpochMs, string Url);
 
 /// <summary>Shared defaults for the changes feed.</summary>
 public static class ChangesDefaults

src/services/Elastic.Documentation.Search/ChangesGateway.cs

@@ -12,46 +12,43 @@ namespace Elastic.Documentation.Search;
 /// <summary>
 /// Elasticsearch gateway for the documentation changes feed.
 /// Queries last_updated > since with search_after cursor pagination.
-/// Uses Point In Time (PIT) for consistent pagination across requests.
+/// Uses a shared Point In Time (PIT) for consistent pagination across requests.
 /// </summary>
-public partial class ChangesGateway(ElasticsearchClientAccessor clientAccessor, ILogger<ChangesGateway> logger)
-	: IChangesGateway
+public partial class ChangesGateway(
+	ElasticsearchClientAccessor clientAccessor,
+	SharedPointInTimeManager pitManager,
+	ILogger<ChangesGateway> logger
+) : IChangesGateway
 {
-	private const string PitKeepAlive = "5m";
-
 	public async Task<ChangesResult> GetChangesAsync(ChangesRequest request, Cancel ctx = default)
 	{
 		var fetchSize = request.PageSize + 1;
 
 		try
 		{
-			var pitId = await ResolvePitId(request.Cursor?.PitId, ctx);
+			var pitId = await pitManager.GetPitIdAsync(ctx);
 
 			var response = await Search(request, pitId, fetchSize, ctx);
 
-			if (!response.IsValidResponse)
+			if (!response.IsValidResponse && IsExpiredPit(response))
 			{
-				if (IsExpiredPit(response) && request.Cursor?.PitId is not null)
-				{
-					LogPitExpired(logger);
-					pitId = await OpenPit(ctx);
-					var updatedCursor = request.Cursor with { PitId = pitId };
-					response = await Search(request with { Cursor = updatedCursor }, pitId, fetchSize, ctx);
-				}
+				LogPitExpired(logger);
+				await pitManager.HandleExpiredPitAsync(pitId, ctx);
+				pitId = await pitManager.GetPitIdAsync(ctx);
+				response = await Search(request, pitId, fetchSize, ctx);
+			}
 
-				if (!response.IsValidResponse)
-				{
-					var reason = response.ElasticsearchServerError?.Error.Reason ?? "Unknown";
-					throw new InvalidOperationException(
-						$"Elasticsearch changes query failed (HTTP {response.ApiCallDetails?.HttpStatusCode}): {reason}"
-					);
-				}
+			if (!response.IsValidResponse)
+			{
+				var reason = response.ElasticsearchServerError?.Error.Reason ?? "Unknown";
+				throw new InvalidOperationException(
+					$"Elasticsearch changes query failed (HTTP {response.ApiCallDetails?.HttpStatusCode}): {reason}"
+				);
 			}
 
-			// Use the PIT ID from the response if available, as ES may return a new one
-			var responsePitId = response.PitId ?? pitId;
+			pitManager.RefreshKeepAlive();
 
-			return BuildResult(response, request.PageSize, responsePitId);
+			return BuildResult(response, request.PageSize);
 		}
 		catch (Exception ex)
 		{
@@ -60,33 +57,6 @@ public async Task<ChangesResult> GetChangesAsync(ChangesRequest request, Cancel
 		}
 	}
 
-	private async Task<string> ResolvePitId(string? existingPitId, Cancel ctx)
-	{
-		if (!string.IsNullOrEmpty(existingPitId))
-			return existingPitId;
-
-		return await OpenPit(ctx);
-	}
-
-	private async Task<string> OpenPit(Cancel ctx)
-	{
-		var response = await clientAccessor.Client.OpenPointInTimeAsync(
-			clientAccessor.SearchIndex,
-			r => r.KeepAlive(PitKeepAlive),
-			ctx
-		);
-
-		if (!response.IsValidResponse)
-		{
-			throw new InvalidOperationException(
-				$"Failed to open PIT: {response.ElasticsearchServerError?.Error.Reason ?? "Unknown"}"
-			);
-		}
-
-		LogPitOpened(logger, response.Id);
-		return response.Id;
-	}
-
 	private async Task<SearchResponse<DocumentationDocument>> Search(
 		ChangesRequest request, string pitId, int fetchSize, Cancel ctx
 	) =>
@@ -95,7 +65,7 @@ await clientAccessor.Client.SearchAsync<DocumentationDocument>(s =>
 			_ = s
 				.Size(fetchSize)
 				.TrackTotalHits(t => t.Enabled(false))
-				.Pit(p => p.Id(pitId).KeepAlive(PitKeepAlive))
+				.Pit(p => p.Id(pitId).KeepAlive(SharedPointInTimeManager.PitKeepAlive))
 				.Query(q => q.Range(r => r
 					.Date(dr => dr
 						.Field(f => f.LastUpdated)
@@ -132,7 +102,7 @@ private static bool IsExpiredPit(SearchResponse<DocumentationDocument> response)
 		|| response.ElasticsearchServerError?.Error.Reason?.Contains("point in time", StringComparison.OrdinalIgnoreCase) == true
 		|| response.ElasticsearchServerError?.Error.Reason?.Contains("No search context found", StringComparison.OrdinalIgnoreCase) == true;
 
-	private static ChangesResult BuildResult(SearchResponse<DocumentationDocument> response, int pageSize, string pitId)
+	private static ChangesResult BuildResult(SearchResponse<DocumentationDocument> response, int pageSize)
 	{
 		var hits = response.Hits.ToList();
 		var hasMore = hits.Count > pageSize;
@@ -167,7 +137,7 @@ private static ChangesResult BuildResult(SearchResponse<DocumentationDocument> r
 					: default(long?);
 
 				if (epochMs is not null && sortUrl.TryGetString(out var url))
-					nextCursor = new ChangesPageCursor(epochMs.Value, url!, pitId);
+					nextCursor = new ChangesPageCursor(epochMs.Value, url!);
 			}
 		}
 
@@ -178,9 +148,6 @@ private static ChangesResult BuildResult(SearchResponse<DocumentationDocument> r
 		};
 	}
 
-	[LoggerMessage(Level = LogLevel.Debug, Message = "Opened new PIT: {PitId}")]
-	private static partial void LogPitOpened(ILogger logger, string pitId);
-
 	[LoggerMessage(Level = LogLevel.Warning, Message = "PIT expired or not found, opening a new one and retrying with existing search_after position")]
 	private static partial void LogPitExpired(ILogger logger);
 }

src/services/Elastic.Documentation.Search/ServicesExtension.cs

@@ -38,6 +38,7 @@ public static IServiceCollection AddSearchServices(this IServiceCollection servi
 		logger?.LogInformation("Full search use case registered with hybrid RRF support");
 
 		// Changes feed (cursor-paginated changes since a given date)
+		_ = services.AddSingleton<SharedPointInTimeManager>();
 		_ = services.AddScoped<IChangesGateway, ChangesGateway>();
 		_ = services.AddScoped<ChangesUsecase>();
 

src/services/Elastic.Documentation.Search/SharedPointInTimeManager.cs

@@ -0,0 +1,111 @@
+// Licensed to Elasticsearch B.V under one or more agreements.
+// Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information
+
+using Elastic.Documentation.Search.Common;
+using Microsoft.Extensions.Logging;
+
+namespace Elastic.Documentation.Search;
+
+/// <summary>Singleton manager for a shared Elasticsearch Point In Time (PIT).</summary>
+public sealed partial class SharedPointInTimeManager(
+	ElasticsearchClientAccessor clientAccessor,
+	ILogger<SharedPointInTimeManager> logger
+) : IAsyncDisposable
+{
+	private static readonly TimeSpan KeepAliveDuration = TimeSpan.FromMinutes(5);
+	public const string PitKeepAlive = "5m";
+
+	private readonly SemaphoreSlim _semaphore = new(1, 1);
+	private string? _pitId;
+	private DateTimeOffset _expiresAt;
+
+	/// <summary>Returns a valid PIT ID, opening a new one if needed.</summary>
+	public async Task<string> GetPitIdAsync(Cancel ctx)
+	{
+		await _semaphore.WaitAsync(ctx);
+		try
+		{
+			if (_pitId is not null && DateTimeOffset.UtcNow < _expiresAt)
+				return _pitId;
+
+			_pitId = await OpenPit(ctx);
+			_expiresAt = DateTimeOffset.UtcNow.Add(KeepAliveDuration);
+			return _pitId;
+		}
+		finally
+		{
+			_ = _semaphore.Release();
+		}
+	}
+
+	/// <summary>Nullifies the current PIT only if it matches the expired one, so a concurrent refresh is not discarded.</summary>
+	public async Task HandleExpiredPitAsync(string expiredPitId, Cancel ctx)
+	{
+		await _semaphore.WaitAsync(ctx);
+		try
+		{
+			if (_pitId == expiredPitId)
+			{
+				LogPitExpired(logger);
+				_pitId = null;
+			}
+		}
+		finally
+		{
+			_ = _semaphore.Release();
+		}
+	}
+
+	/// <summary>Bumps the local expiry after a successful search (ES extends the PIT server-side via KeepAlive).</summary>
+	public void RefreshKeepAlive() => _expiresAt = DateTimeOffset.UtcNow.Add(KeepAliveDuration);
+
+	private async Task<string> OpenPit(Cancel ctx)
+	{
+		var response = await clientAccessor.Client.OpenPointInTimeAsync(
+			clientAccessor.SearchIndex,
+			r => r.KeepAlive(PitKeepAlive),
+			ctx
+		);
+
+		if (!response.IsValidResponse)
+		{
+			throw new InvalidOperationException(
+				$"Failed to open PIT: {response.ElasticsearchServerError?.Error.Reason ?? "Unknown"}"
+			);
+		}
+
+		LogPitOpened(logger, response.Id);
+		return response.Id;
+	}
+
+	/// <inheritdoc />
+	public async ValueTask DisposeAsync()
+	{
+		if (_pitId is null)
+			return;
+
+		try
+		{
+			_ = await clientAccessor.Client.ClosePointInTimeAsync(r => r.Id(_pitId));
+			LogPitClosed(logger, _pitId);
+		}
+		catch (Exception ex)
+		{
+			logger.LogWarning(ex, "Failed to close PIT {PitId} during shutdown", _pitId);
+		}
+		finally
+		{
+			_semaphore.Dispose();
+		}
+	}
+
+	[LoggerMessage(Level = LogLevel.Debug, Message = "Opened new shared PIT: {PitId}")]
+	private static partial void LogPitOpened(ILogger logger, string pitId);
+
+	[LoggerMessage(Level = LogLevel.Warning, Message = "Shared PIT expired or not found, will open a new one")]
+	private static partial void LogPitExpired(ILogger logger);
+
+	[LoggerMessage(Level = LogLevel.Debug, Message = "Closed shared PIT: {PitId}")]
+	private static partial void LogPitClosed(ILogger logger, string pitId);
+}