From 9daa71fb0dbfaa9bf0484c6d6e2128f71d6d58bc Mon Sep 17 00:00:00 2001 From: Felipe Cotti Date: Tue, 28 Jul 2026 11:19:54 -0300 Subject: [PATCH 1/3] Scrubber Lambda owns the public changelog registry via state reconcile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of elastic/docs-eng-team#688. The public registry.json was a log of upload operations (client-written, pass-through copied); every known consistency gap followed from that. The scrubber Lambda now derives it from the public bucket's actual state: registry = f(state), never f(event). - Extract the Lambda's top-level handler logic into testable classes in Elastic.Changelog: ScrubberProcessor (batch coalescing by key and group, object-level reconcile with post-write source validation) and RegistryReconciler (delimited/paginated group listing, ETag reuse with amends always recomputed, semantic idempotence, conditional PUT/DELETE with bounded jittered retries on 412/409, newer-schema refusal). Program.cs is now a thin adapter. - Retire the registry pass-through in the same deploy: registry-key events only schedule a group reconcile, so client-authored JSON no longer reaches the public bucket uninspected. - Add a producer (algorithm version) field to the manifest; a mismatch — including legacy pass-through manifests — forces a full metadata recompute and a write even when entries are identical. - Emit per-invocation reconcile metrics as CloudWatch EMF (the Phase 0 observability item that could only land with the reconciler). Co-Authored-By: Claude Fable 5 --- .../EmfMetricsEmitter.cs | 77 +++ .../LambdaLoggerFactory.cs | 57 +++ .../docs-lambda-changelog-scrubber/Program.cs | 222 +-------- .../Reconciliation/ChangelogScope.cs | 92 ++++ .../Reconciliation/ReconcileMetrics.cs | 66 +++ .../Reconciliation/RegistryReconciler.cs | 438 +++++++++++++++++ .../Scrubbing/ChangelogContentScrubber.cs | 117 +++++ .../Scrubbing/ScrubberProcessor.cs | 285 +++++++++++ .../Elastic.Changelog/Uploading/Registry.cs | 13 +- .../Reconciliation/FakeS3.cs | 231 +++++++++ .../Reconciliation/RegistryReconcilerTests.cs | 463 ++++++++++++++++++ .../Scrubbing/ScrubberProcessorTests.cs | 286 +++++++++++ 12 files changed, 2140 insertions(+), 207 deletions(-) create mode 100644 src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs create mode 100644 src/infra/docs-lambda-changelog-scrubber/LambdaLoggerFactory.cs create mode 100644 src/services/Elastic.Changelog/Reconciliation/ChangelogScope.cs create mode 100644 src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs create mode 100644 src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs create mode 100644 src/services/Elastic.Changelog/Scrubbing/ChangelogContentScrubber.cs create mode 100644 src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs create mode 100644 tests/Elastic.Changelog.Tests/Reconciliation/FakeS3.cs create mode 100644 tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs create mode 100644 tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs diff --git a/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs b/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs new file mode 100644 index 0000000000..ffb34436f7 --- /dev/null +++ b/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs @@ -0,0 +1,77 @@ +// 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 System.Text; +using System.Text.Json; +using Elastic.Changelog.Reconciliation; + +namespace Elastic.Documentation.Lambda.ChangelogScrubber; + +/// +/// Emits the per-invocation reconcile counters as a CloudWatch Embedded Metric Format line +/// (elastic/docs-eng-team#688 Phase 0 observability: these numbers gate any later SQS/Lambda +/// tuning). Written with directly — no serializer registration, so +/// nothing for AOT trimming to miss — and to stdout unwrapped, which is what the EMF parser +/// requires. +/// +internal static class EmfMetricsEmitter +{ + private const string Namespace = "docs-changelog-scrubber"; + + private static readonly (string Name, Func Value)[] Counters = + [ + ("ObjectReconciles", m => m.ObjectReconciles), + ("ObjectReconcileRetries", m => m.ObjectReconcileRetries), + ("GroupReconciles", m => m.GroupReconciles), + ("RegistryWrites", m => m.RegistryWrites), + ("RegistryDeletes", m => m.RegistryDeletes), + ("RegistryUnchanged", m => m.RegistryUnchanged), + ("WriteConflicts", m => m.WriteConflicts), + ("ObjectsListed", m => m.ObjectsListed), + ("EntriesRecomputed", m => m.EntriesRecomputed), + ("FailedMessages", m => m.FailedMessages) + ]; + + public static void Emit(ReconcileMetrics metrics) + { + using var buffer = new MemoryStream(); + using (var writer = new Utf8JsonWriter(buffer)) + { + writer.WriteStartObject(); + + writer.WritePropertyName("_aws"); + writer.WriteStartObject(); + writer.WriteNumber("Timestamp", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()); + writer.WritePropertyName("CloudWatchMetrics"); + writer.WriteStartArray(); + writer.WriteStartObject(); + writer.WriteString("Namespace", Namespace); + writer.WritePropertyName("Dimensions"); + writer.WriteStartArray(); + writer.WriteStartArray(); + writer.WriteEndArray(); + writer.WriteEndArray(); + writer.WritePropertyName("Metrics"); + writer.WriteStartArray(); + foreach (var (name, _) in Counters) + { + writer.WriteStartObject(); + writer.WriteString("Name", name); + writer.WriteString("Unit", "Count"); + writer.WriteEndObject(); + } + writer.WriteEndArray(); + writer.WriteEndObject(); + writer.WriteEndArray(); + writer.WriteEndObject(); + + foreach (var (name, value) in Counters) + writer.WriteNumber(name, value(metrics)); + + writer.WriteEndObject(); + } + + Console.WriteLine(Encoding.UTF8.GetString(buffer.ToArray())); + } +} diff --git a/src/infra/docs-lambda-changelog-scrubber/LambdaLoggerFactory.cs b/src/infra/docs-lambda-changelog-scrubber/LambdaLoggerFactory.cs new file mode 100644 index 0000000000..69ae19ab78 --- /dev/null +++ b/src/infra/docs-lambda-changelog-scrubber/LambdaLoggerFactory.cs @@ -0,0 +1,57 @@ +// 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 Amazon.Lambda.Core; +using Microsoft.Extensions.Logging; +using LogLevel = Microsoft.Extensions.Logging.LogLevel; + +namespace Elastic.Documentation.Lambda.ChangelogScrubber; + +/// +/// Routes calls from the shared Elastic.Changelog processing code to the +/// Lambda's own logger, so the extracted processor stays free of Lambda dependencies. +/// +internal sealed class LambdaLoggerFactory(ILambdaLogger lambdaLogger) : ILoggerFactory +{ + public ILogger CreateLogger(string categoryName) => new LambdaLoggerAdapter(categoryName, lambdaLogger); + + public void AddProvider(ILoggerProvider provider) + { + // Providers are meaningless here; everything goes to the Lambda logger. + } + + public void Dispose() + { + } + + private sealed class LambdaLoggerAdapter(string categoryName, ILambdaLogger lambdaLogger) : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Information; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + if (!IsEnabled(logLevel)) + return; + + var message = $"{categoryName}: {formatter(state, exception)}"; + if (exception is not null) + message = $"{message} | {exception}"; + + lambdaLogger.Log(MapLevel(logLevel), message); + } + + private static string MapLevel(LogLevel level) => level switch + { + LogLevel.Trace => "TRACE", + LogLevel.Debug => "DEBUG", + LogLevel.Information => "INFO", + LogLevel.Warning => "WARN", + LogLevel.Error => "ERROR", + LogLevel.Critical => "CRITICAL", + _ => "INFO" + }; + } +} diff --git a/src/infra/docs-lambda-changelog-scrubber/Program.cs b/src/infra/docs-lambda-changelog-scrubber/Program.cs index 350dd3a0a5..c1f668129c 100644 --- a/src/infra/docs-lambda-changelog-scrubber/Program.cs +++ b/src/infra/docs-lambda-changelog-scrubber/Program.cs @@ -2,22 +2,17 @@ // 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 System.Net; using System.Reflection; -using System.Text.Json; using Amazon.Lambda.Core; using Amazon.Lambda.RuntimeSupport; using Amazon.Lambda.Serialization.SystemTextJson; using Amazon.Lambda.SQSEvents; using Amazon.S3; -using Amazon.S3.Model; -using Amazon.S3.Util; using Elastic.Changelog.Bundling; +using Elastic.Changelog.Reconciliation; +using Elastic.Changelog.Scrubbing; using Elastic.Documentation.Configuration.Assembler; -using Elastic.Documentation.Configuration.ReleaseNotes; -using Elastic.Documentation.Diagnostics; using Elastic.Documentation.Lambda.ChangelogScrubber; -using Elastic.Documentation.ReleaseNotes; var publicBucketName = Environment.GetEnvironmentVariable("PUBLIC_BUCKET_NAME") ?? throw new InvalidOperationException("PUBLIC_BUCKET_NAME environment variable is required"); @@ -41,6 +36,8 @@ IReadOnlyList BuildAllowlist() return LinkAllowlistSanitizer.BuildAllowReposFromAssembler(assembly); } +// Thin adapter over the testable processor in Elastic.Changelog: translate the SQS event in, +// run the state-driven reconcile, translate the failed message ids back out, emit metrics. async Task Handler(SQSEvent ev, ILambdaContext context) { var region = Amazon.RegionEndpoint.GetBySystemName( @@ -54,207 +51,20 @@ async Task Handler(SQSEvent ev, ILambdaContext context) MaxErrorRetry = 2 }); - var batchItemFailures = new List(); + using var logFactory = new LambdaLoggerFactory(context.Logger); + var metrics = new ReconcileMetrics(); + var scrubber = new ChangelogContentScrubber(logFactory, allowRepos); + var reconciler = new RegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); + var processor = new ScrubberProcessor(logFactory, s3Client, publicBucketName, scrubber, reconciler, metrics); - foreach (var message in ev.Records) - { - try - { - var s3Event = S3EventNotification.ParseJson(message.Body); - foreach (var record in s3Event.Records) - { - var key = Uri.UnescapeDataString(record.S3.Object.Key.Replace('+', ' ')); - var sourceBucket = record.S3.Bucket.Name; - var eventName = record.EventName; - - context.Logger.LogInformation("Processing event={EventName} key={Key}", eventName, key); + var messages = ev.Records.Select(r => new ScrubberQueueMessage(r.MessageId, r.Body)).ToList(); + var failedIds = await processor.ProcessAsync(messages, CancellationToken.None); - if (eventName.Value.Contains("ObjectRemoved")) - { - await DeleteFromPublicBucket(s3Client, key, context); - } - else if (eventName.Value.Contains("ObjectCreated")) - { - await ScrubAndCopyToPublicBucket(s3Client, sourceBucket, key, context); - } - else - { - context.Logger.LogWarning("Ignoring unhandled event type: {EventName}", eventName); - } - } - } - catch (Exception e) - { - context.Logger.LogWarning(e, "Failed to process message {MessageId}", message.MessageId); - batchItemFailures.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = message.MessageId }); - } - } + EmfMetricsEmitter.Emit(metrics); - var response = new SQSBatchResponse(batchItemFailures); - if (batchItemFailures.Count > 0) - context.Logger.LogInformation("Failed {FailedCount} of {TotalCount} messages", batchItemFailures.Count, ev.Records.Count); - - var jsonStr = JsonSerializer.Serialize(response, SerializerContext.Default.SQSBatchResponse); - context.Logger.LogInformation(jsonStr); + var response = new SQSBatchResponse( + [.. failedIds.Select(id => new SQSBatchResponse.BatchItemFailure { ItemIdentifier = id })]); + if (failedIds.Count > 0) + context.Logger.LogInformation("Failed {FailedCount} of {TotalCount} messages", failedIds.Count, ev.Records.Count); return response; } - -async Task DeleteFromPublicBucket(IAmazonS3 s3Client, string key, ILambdaContext context) -{ - context.Logger.LogDebug($"Removing {key} from the public bucket", key); - try - { - _ = await s3Client.DeleteObjectAsync(new DeleteObjectRequest - { - BucketName = publicBucketName, - Key = key - }); - context.Logger.LogInformation("Deleted {Key} from public bucket", key); - } - catch (AmazonS3Exception e) when (e.StatusCode == HttpStatusCode.NotFound) - { - context.Logger.LogInformation("Key {Key} already absent from public bucket", key); - } -} - -async Task ScrubAndCopyToPublicBucket(IAmazonS3 s3Client, string sourceBucket, string key, ILambdaContext context) -{ - context.Logger.LogDebug("Scrubbing {Key} to public bucket", key); - - if (ChangelogKeys.IsRegistry(key)) - { - await CopyPassThrough(s3Client, sourceBucket, key, context); - return; - } - - if (key.EndsWith(".json", StringComparison.OrdinalIgnoreCase)) - { - context.Logger.LogWarning("Skipping unapproved JSON key: {Key}", key); - return; - } - - if (!key.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase) && - !key.EndsWith(".yml", StringComparison.OrdinalIgnoreCase)) - { - context.Logger.LogInformation("Skipping non-YAML key: {Key}", key); - return; - } - - context.Logger.LogInformation("Getting {Key} from private bucket", key); - var getResponse = await s3Client.GetObjectAsync(new GetObjectRequest - { - BucketName = sourceBucket, - Key = key - }); - - string content; - using (var reader = new StreamReader(getResponse.ResponseStream)) - content = await reader.ReadToEndAsync(); - - context.Logger.LogInformation("Performing scrub pass for {Key}", key); - var scrubbed = await ScrubContent(key, content, context); - - context.Logger.LogInformation("Putting scrubbed {Key} on public bucket", key); - _ = await s3Client.PutObjectAsync(new PutObjectRequest - { - BucketName = publicBucketName, - Key = key, - ContentBody = scrubbed, - ContentType = "application/yaml" - }); - - context.Logger.LogInformation("Scrubbed and wrote {Key} to public bucket", key); -} - -async Task CopyPassThrough(IAmazonS3 s3Client, string sourceBucket, string key, ILambdaContext context) -{ - _ = await s3Client.CopyObjectAsync(new CopyObjectRequest - { - SourceBucket = sourceBucket, - SourceKey = key, - DestinationBucket = publicBucketName, - DestinationKey = key - }); - context.Logger.LogInformation("Copied {Key} to public bucket (pass-through)", key); -} - -async Task ScrubContent(string key, string content, ILambdaContext context) -{ - // Artifact-root layout: bundles live under "bundle/{product}/...", entries under - // "changelog/{org}/{repo}/{branch}/...". Match the bundle prefix (not a "/bundle/" substring, which no - // longer appears in the new keys) so bundles are not misclassified as changelog entries. - var isBundlePath = key.StartsWith(ChangelogKeys.BundlePrefix, StringComparison.OrdinalIgnoreCase); - - if (isBundlePath) - return await ScrubBundle(content, context); - - return await ScrubChangelog(content, context); -} - -async Task ScrubBundle(string content, ILambdaContext context) -{ - var bundle = ReleaseNotesSerialization.DeserializeBundle(content); - var owner = bundle.Products.Count > 0 ? bundle.Products[0].Owner ?? "elastic" : "elastic"; - var repo = bundle.Products.Count > 0 ? bundle.Products[0].Repo : null; - - await using var collector = new DiagnosticsCollector([]); - if (!LinkAllowlistSanitizer.ScrubBundleForPublic(collector, bundle, allowRepos, owner, repo, out var sanitized, out var changed)) - throw new InvalidOperationException($"Failed to scrub bundle for public output; errors: {collector.Errors}"); - - if (!changed) - { - context.Logger.LogInformation("Bundle had no private references, writing unchanged"); - LinkAllowlistSanitizer.ValidateNoPrivateReferences(content, allowRepos); - return content; - } - - var result = ReleaseNotesSerialization.SerializeBundle(sanitized); - LinkAllowlistSanitizer.ValidateNoPrivateReferences(result, allowRepos); - return result; -} - -async Task ScrubChangelog(string content, ILambdaContext context) -{ - var normalized = ReleaseNotesSerialization.NormalizeYaml(content); - var entry = ReleaseNotesSerialization.DeserializeEntry(normalized); - - var bundledEntry = new BundledEntry - { - Type = entry.Type, - Title = entry.Title, - Description = entry.Description, - Impact = entry.Impact, - Action = entry.Action, - Prs = entry.Prs, - Issues = entry.Issues, - Areas = entry.Areas, - Highlight = entry.Highlight, - Subtype = entry.Subtype - }; - - await using var collector = new DiagnosticsCollector([]); - if (!LinkAllowlistSanitizer.TryApplyChangelogEntry( - collector, bundledEntry, allowRepos, "elastic", null, - out var sanitized, out var changed)) - throw new InvalidOperationException($"Failed to apply allowlist to changelog entry; errors: {collector.Errors}"); - - if (!changed) - { - context.Logger.LogInformation("Changelog entry had no private references, writing unchanged"); - LinkAllowlistSanitizer.ValidateNoPrivateReferences(content, allowRepos); - return content; - } - - var scrubEntry = entry with - { - Description = sanitized.Description, - Impact = sanitized.Impact, - Action = sanitized.Action, - Prs = sanitized.Prs, - Issues = sanitized.Issues - }; - - var result = ReleaseNotesSerialization.SerializeEntry(scrubEntry); - LinkAllowlistSanitizer.ValidateNoPrivateReferences(result, allowRepos); - return result; -} diff --git a/src/services/Elastic.Changelog/Reconciliation/ChangelogScope.cs b/src/services/Elastic.Changelog/Reconciliation/ChangelogScope.cs new file mode 100644 index 0000000000..c81da6fba0 --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ChangelogScope.cs @@ -0,0 +1,92 @@ +// 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 System.Diagnostics.CodeAnalysis; +using System.Text.Json.Serialization; +using Elastic.Documentation.Configuration.ReleaseNotes; + +namespace Elastic.Changelog.Reconciliation; + +/// The two registry scope families in the changelog bucket key layout. +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum ChangelogScopeKind +{ + /// A product bundle scope: bundle/{product}/…. + Bundle, + + /// An authoring-pool scope: changelog/{org}/{repo}/{branch}/…. + Changelog +} + +/// +/// Identifies one registry scope in the changelog bundles bucket — a product bundle pool +/// (bundle/{product}/) or an authoring changelog pool +/// (changelog/{org}/{repo}/{branch}/) — and derives the scope's key prefix and +/// registry.json key. Segments are validated on construction via +/// , so a scope instance can always be composed into safe S3 keys. +/// +public sealed record ChangelogScope +{ + private ChangelogScope(ChangelogScopeKind kind, string group) + { + Kind = kind; + Group = group; + } + + /// Which scope family this is. + public ChangelogScopeKind Kind { get; } + + /// + /// The grouping segment(s): the product for a bundle scope, the + /// {org}/{repo}/{branch} prefix for a changelog scope. + /// + public string Group { get; } + + /// The S3 key prefix of every object in this scope, ending in /. + public string Prefix => Kind == ChangelogScopeKind.Bundle + ? $"{ChangelogKeys.BundlePrefix}{Group}/" + : $"{ChangelogKeys.ChangelogPrefix}{Group}/"; + + /// The S3 key of this scope's registry.json manifest. + public string RegistryKey => Kind == ChangelogScopeKind.Bundle + ? ChangelogKeys.BundleRegistryKey(Group) + : ChangelogKeys.ChangelogRegistryKey(Group); + + /// Creates a bundle scope for ; false when the segment is invalid. + public static bool TryCreateBundle(string? product, [NotNullWhen(true)] out ChangelogScope? scope) + { + scope = ChangelogKeys.IsValidProduct(product) + ? new ChangelogScope(ChangelogScopeKind.Bundle, product) + : null; + return scope is not null; + } + + /// Creates a changelog-pool scope for //; false when any segment is invalid. + public static bool TryCreateChangelog(string? org, string? repo, string? branch, [NotNullWhen(true)] out ChangelogScope? scope) + { + scope = ChangelogKeys.IsValidOrg(org) && ChangelogKeys.IsValidRepo(repo) && ChangelogKeys.IsValidBranch(branch) + ? new ChangelogScope(ChangelogScopeKind.Changelog, $"{org}/{repo}/{branch}") + : null; + return scope is not null; + } + + /// + /// Derives the scope an object key belongs to — bundle/{product}/{file} or + /// changelog/{org}/{repo}/{branch}/{file}, including the scope's own + /// registry.json key. False when the key sits outside both layouts or a segment + /// fails validation. + /// + public static bool TryFromKey(string key, [NotNullWhen(true)] out ChangelogScope? scope) + { + scope = null; + if (ChangelogKeys.ExtractBundleGroup(key) is { } product) + scope = new ChangelogScope(ChangelogScopeKind.Bundle, product); + else if (ChangelogKeys.ExtractChangelogGroup(key) is { } pool) + scope = new ChangelogScope(ChangelogScopeKind.Changelog, pool); + return scope is not null; + } + + /// + public override string ToString() => Prefix.TrimEnd('/'); +} diff --git a/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs b/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs new file mode 100644 index 0000000000..4d6eb2c70f --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs @@ -0,0 +1,66 @@ +// 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 + +namespace Elastic.Changelog.Reconciliation; + +/// +/// Per-invocation counters for the scrubber pipeline (elastic/docs-eng-team#688 Phase 0 +/// observability): once the Lambda owns the public registry, these numbers are what gates any +/// SQS/Lambda tuning — first-heal cost, conditional-write contention, steady-state reuse rate. +/// Thread-safe: entry recomputes run under bounded parallelism. +/// +public sealed class ReconcileMetrics +{ + private int _objectReconciles; + private int _objectReconcileRetries; + private int _groupReconciles; + private int _registryWrites; + private int _registryDeletes; + private int _registryUnchanged; + private int _writeConflicts; + private int _objectsListed; + private int _entriesRecomputed; + private int _failedMessages; + + /// Object-level reconciles run (one per distinct key in the batch). + public int ObjectReconciles => _objectReconciles; + + /// Object reconciles redone because the private source changed mid-flight (post-write validation). + public int ObjectReconcileRetries => _objectReconcileRetries; + + /// Group-level reconciles run (one per distinct group in the batch). + public int GroupReconciles => _groupReconciles; + + /// Public manifests written. + public int RegistryWrites => _registryWrites; + + /// Public manifests deleted (empty groups). + public int RegistryDeletes => _registryDeletes; + + /// Group reconciles that found the manifest already exact (steady state). + public int RegistryUnchanged => _registryUnchanged; + + /// Conditional writes (PUT or DELETE) lost to a concurrent writer. + public int WriteConflicts => _writeConflicts; + + /// Objects seen across all group listings. + public int ObjectsListed => _objectsListed; + + /// Entries whose metadata was recomputed from a public YAML read (vs. ETag-reused). + public int EntriesRecomputed => _entriesRecomputed; + + /// SQS messages reported as batch-item failures. + public int FailedMessages => _failedMessages; + + internal void IncrementObjectReconciles() => Interlocked.Increment(ref _objectReconciles); + internal void IncrementObjectReconcileRetries() => Interlocked.Increment(ref _objectReconcileRetries); + internal void IncrementGroupReconciles() => Interlocked.Increment(ref _groupReconciles); + internal void IncrementRegistryWrites() => Interlocked.Increment(ref _registryWrites); + internal void IncrementRegistryDeletes() => Interlocked.Increment(ref _registryDeletes); + internal void IncrementRegistryUnchanged() => Interlocked.Increment(ref _registryUnchanged); + internal void IncrementWriteConflicts() => Interlocked.Increment(ref _writeConflicts); + internal void IncrementObjectsListed() => Interlocked.Increment(ref _objectsListed); + internal void IncrementEntriesRecomputed() => Interlocked.Increment(ref _entriesRecomputed); + internal void AddFailedMessages(int count) => Interlocked.Add(ref _failedMessages, count); +} diff --git a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs b/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs new file mode 100644 index 0000000000..a1148ae8f9 --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs @@ -0,0 +1,438 @@ +// 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 System.Net; +using System.Text.Json; +using Amazon.S3; +using Amazon.S3.Model; +using Elastic.Changelog.Uploading; +using Elastic.Documentation.Configuration.ReleaseNotes; +using Elastic.Documentation.ReleaseNotes; +using Elastic.Documentation.Versions; +using Microsoft.Extensions.Logging; + +namespace Elastic.Changelog.Reconciliation; + +/// How a group reconcile converged. +public enum GroupReconcileOutcome +{ + /// No public objects and no manifest — nothing to do. + NoOp, + + /// The existing manifest already describes the listing exactly; no write issued. + Unchanged, + + /// The manifest was (re)written from the public listing. + Written, + + /// The group is empty; its manifest was conditionally deleted. + Deleted, + + /// The existing manifest declares a newer schema than this producer understands; left untouched. + RefusedNewerSchema +} + +/// A conditional registry write kept losing races after every retry; the SQS message should be redelivered. +public sealed class ReconcileConflictException(string message) : Exception(message); + +/// +/// Rebuilds one group's public registry.json from the current public bucket state +/// (registry = f(state), never f(event) — see elastic/docs-eng-team#688). Lists the +/// group's prefix, reuses entries whose recorded ETag still matches, recomputes the rest from the +/// scrubbed public YAMLs, and writes the manifest back with optimistic concurrency. Any successful +/// reconcile therefore repairs all accumulated drift in the group, not just the change +/// that triggered it. +/// +public sealed class RegistryReconciler( + ILoggerFactory logFactory, + IAmazonS3 s3Client, + string publicBucketName, + TimeProvider? timeProvider = null, + TimeSpan? retryBaseDelay = null, + ReconcileMetrics? metrics = null +) +{ + /// + /// Identifies this reconciliation algorithm version in written manifests. Bump on any change to + /// how entries are computed: a mismatch forces a full recompute (and a write even when the + /// entries come out identical), which is how metadata-logic fixes roll out to every group. + /// + public const string Producer = "changelog-scrubber-reconcile/1"; + + // Bounds the optimistic-concurrency retry loop; each attempt re-lists and re-reads before retrying. + private const int MaxWriteAttempts = 5; + + // First-heal of a large group GETs every unlisted YAML; keep those reads bounded. + private const int MaxParallelReads = 4; + + private readonly ILogger _logger = logFactory.CreateLogger(); + private readonly TimeProvider _time = timeProvider ?? TimeProvider.System; + private readonly TimeSpan _retryBaseDelay = retryBaseDelay ?? TimeSpan.FromMilliseconds(200); + private readonly ReconcileMetrics _metrics = metrics ?? new ReconcileMetrics(); + + /// + /// Converges the group's public manifest to its public listing. Throws + /// when concurrent conditional writers win every + /// bounded retry — the caller fails the SQS message so redelivery retries later. + /// + public async Task ReconcileGroupAsync(ChangelogScope scope, Cancel ctx) + { + _metrics.IncrementGroupReconciles(); + + for (var attempt = 1; attempt <= MaxWriteAttempts; attempt++) + { + ctx.ThrowIfCancellationRequested(); + + var listing = await ListGroupFiles(scope, ctx); + var existing = await FetchManifest(scope.RegistryKey, ctx); + + // Empty group: conditionally delete before any equality check, so a stale empty + // observation cannot destroy a concurrent reconciler's fresh manifest, and a manifest + // whose bundles are already [] is still removed rather than short-circuited. Absent ≠ + // empty for consumers: deleting restores "unpublished" (404) semantics for the group. + if (listing.Count == 0) + { + if (!existing.Exists) + return GroupReconcileOutcome.NoOp; + + if (await TryDeleteManifest(scope, existing.ETag!, attempt, ctx)) + return GroupReconcileOutcome.Deleted; + await BackOff(attempt, ctx); + continue; + } + + // Never rewrite (and implicitly downgrade) a manifest produced by a newer schema. + if (existing.Manifest is { } newer && newer.SchemaVersion > Registry.CurrentSchemaVersion) + { + _logger.LogWarning( + "Public manifest {Key} declares schema_version {Found} > supported {Supported}; leaving it untouched", + scope.RegistryKey, newer.SchemaVersion, Registry.CurrentSchemaVersion); + return GroupReconcileOutcome.RefusedNewerSchema; + } + + // Entries are only reusable — and the write only skippable — when the whole manifest + // is trustworthy: parsed cleanly and produced by this algorithm for this group. A + // corrupt manifest or a producer/schema/product mismatch forces a full recompute and + // a write even when the entries come out identical, otherwise the producer version + // would never be adopted and every future reconcile would keep recomputing. + var trusted = existing is { Manifest: not null, Corrupt: false } + && existing.Manifest.SchemaVersion == Registry.CurrentSchemaVersion + && string.Equals(existing.Manifest.Producer, Producer, StringComparison.Ordinal) + && string.Equals(existing.Manifest.Product, scope.Group, StringComparison.Ordinal); + + var (entries, reused) = await BuildEntries(scope, listing, trusted ? existing.Manifest!.Bundles : [], ctx); + + if (trusted && BundlesEqual(existing.Manifest!.Bundles, entries)) + { + _metrics.IncrementRegistryUnchanged(); + _logger.LogDebug("Public manifest {Key} already matches the listing; skipping write", scope.RegistryKey); + return GroupReconcileOutcome.Unchanged; + } + + var manifest = new Registry + { + Product = scope.Group, + Producer = Producer, + GeneratedAt = _time.GetUtcNow(), + Bundles = entries + }; + var json = JsonSerializer.Serialize(manifest, RegistryJsonContext.Default.Registry); + + if (await TryPutManifest(scope, json, existing.ETag, attempt, ctx)) + { + _metrics.IncrementRegistryWrites(); + _logger.LogInformation( + "Wrote public manifest {Key} with {Count} entrie(s) ({Reused} reused, {Recomputed} recomputed)", + scope.RegistryKey, entries.Count, reused, entries.Count - reused); + return GroupReconcileOutcome.Written; + } + await BackOff(attempt, ctx); + } + + throw new ReconcileConflictException( + $"Public manifest {scope.RegistryKey} kept changing concurrently after {MaxWriteAttempts} attempts; failing the message for redelivery."); + } + + /// + /// The group's immediate .yaml/.yml children in the public bucket. The + /// / delimiter matters: branches are stored verbatim, so without it + /// changelog/{org}/{repo}/main/ would also sweep in the main/feature/… pool. + /// + private async Task> ListGroupFiles(ChangelogScope scope, Cancel ctx) + { + var request = new ListObjectsV2Request + { + BucketName = publicBucketName, + Prefix = scope.Prefix, + Delimiter = "/" + }; + + var files = new List(); + ListObjectsV2Response response; + do + { + response = await s3Client.ListObjectsV2Async(request, ctx); + foreach (var obj in response.S3Objects ?? []) + { + var file = obj.Key[scope.Prefix.Length..]; + if (!IsYamlFileName(file) || string.Equals(file, ChangelogKeys.RegistryFileName, StringComparison.Ordinal)) + continue; + files.Add(obj); + _metrics.IncrementObjectsListed(); + } + request.ContinuationToken = response.NextContinuationToken; + } while (response.IsTruncated == true); + + return files; + } + + private static bool IsYamlFileName(string file) => + file.Length > 0 + && (file.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase) || file.EndsWith(".yml", StringComparison.OrdinalIgnoreCase)); + + private sealed record ManifestState(Registry? Manifest, string? ETag, bool Exists, bool Corrupt); + + /// Reads the manifest, distinguishing absent (no ETag) from corrupt (live ETag, no parse). + private async Task FetchManifest(string key, Cancel ctx) + { + string? etag = null; + try + { + using var response = await s3Client.GetObjectAsync(new GetObjectRequest + { + BucketName = publicBucketName, + Key = key + }, ctx); + + etag = response.ETag; + await using var stream = response.ResponseStream; + var manifest = await JsonSerializer.DeserializeAsync(stream, RegistryJsonContext.Default.Registry, ctx); + return new ManifestState(manifest, etag, Exists: true, Corrupt: manifest is null); + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return new ManifestState(null, null, Exists: false, Corrupt: false); + } + catch (JsonException ex) + { + // A corrupt manifest is rebuilt from the listing; its live ETag lets the conditional + // write replace it safely. Transient S3/IO errors bubble up instead — they must not be + // mistaken for corruption. + _logger.LogWarning(ex, "Public manifest {Key} could not be parsed; rebuilding from the listing", key); + return new ManifestState(null, etag, Exists: true, Corrupt: true); + } + } + + private async Task<(List Entries, int Reused)> BuildEntries( + ChangelogScope scope, + IReadOnlyList listing, + IReadOnlyList reusable, + Cancel ctx) + { + // The entry index is listing-only by design (target is null; consumers re-read each entry), + // so the whole build is zero GETs. + if (scope.Kind == ChangelogScopeKind.Changelog) + { + var entries = Sort(listing.Select(obj => new RegistryBundle + { + File = obj.Key[scope.Prefix.Length..], + Target = null, + ETag = NormalizeETag(obj.ETag) + })); + return (entries, entries.Count); + } + + var byFile = reusable.ToDictionary(b => b.File, b => b, StringComparer.Ordinal); + var built = new RegistryBundle?[listing.Count]; + var reused = 0; + + await Parallel.ForEachAsync( + Enumerable.Range(0, listing.Count), + new ParallelOptions { MaxDegreeOfParallelism = MaxParallelReads, CancellationToken = ctx }, + async (i, ct) => + { + var obj = listing[i]; + var file = obj.Key[scope.Prefix.Length..]; + var etag = NormalizeETag(obj.ETag); + + // Amends are never ETag-skipped: their target depends on the parent bundle too, + // and a parent appearing or changing does not touch the amend's own ETag. + if (!BundleAmendMerger.IsAmendFile(file) + && byFile.TryGetValue(file, out var previous) + && string.Equals(previous.ETag, etag, StringComparison.Ordinal)) + { + built[i] = previous; + _ = Interlocked.Increment(ref reused); + return; + } + + var target = await ComputeTarget(scope, file, ct); + _metrics.IncrementEntriesRecomputed(); + built[i] = new RegistryBundle { File = file, Target = target, ETag = etag }; + }); + + // A null slot means the object vanished between the listing and the read; the delete's own + // event (or the next reconcile) covers it. + return (Sort(built.Where(b => b is not null)!), reused); + } + + /// + /// Reads the scrubbed public YAML and extracts the target for the group's product — matching + /// the product id, never blindly the first product. Amends without products inherit the parent + /// bundle's target; an absent parent yields a null target and a warning, self-correcting once + /// the parent lands (never a permanent error, which would block the whole group). + /// + private async Task ComputeTarget(ChangelogScope scope, string file, Cancel ctx) + { + var bundle = await TryReadBundle(scope, file, ctx); + if (bundle is null) + return null; + + if (bundle.Products.Count > 0) + return TargetForProduct(bundle, scope.Group); + + var parentFile = BundleAmendMerger.GetParentBundlePath(file); + if (parentFile is null) + return null; + + var parent = await TryReadBundle(scope, parentFile, ctx); + if (parent is null || parent.Products.Count == 0) + { + _logger.LogWarning( + "Amend {Prefix}{File} has no parent bundle {Parent} in the public bucket yet; recording a null target", + scope.Prefix, file, parentFile); + return null; + } + + return TargetForProduct(parent, scope.Group); + } + + private async Task TryReadBundle(ChangelogScope scope, string file, Cancel ctx) + { + try + { + using var response = await s3Client.GetObjectAsync(new GetObjectRequest + { + BucketName = publicBucketName, + Key = scope.Prefix + file + }, ctx); + + await using var stream = response.ResponseStream; + using var reader = new StreamReader(stream); + var content = await reader.ReadToEndAsync(ctx); + return ReleaseNotesSerialization.DeserializeBundle(content); + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return null; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogWarning(ex, "Could not read bundle target from {Prefix}{File}; recording a null target", scope.Prefix, file); + return null; + } + } + + private static string? TargetForProduct(Bundle bundle, string product) + { + var match = bundle.Products.FirstOrDefault(p => string.Equals(p.ProductId, product, StringComparison.Ordinal)); + return (match ?? bundle.Products[0]).Target; + } + + private async Task TryDeleteManifest(ChangelogScope scope, string etag, int attempt, Cancel ctx) + { + try + { + _ = await s3Client.DeleteObjectAsync(new DeleteObjectRequest + { + BucketName = publicBucketName, + Key = scope.RegistryKey, + IfMatch = etag + }, ctx); + _metrics.IncrementRegistryDeletes(); + _logger.LogInformation("Deleted public manifest {Key}: the group is empty", scope.RegistryKey); + return true; + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + // Another reconciler removed it first; converged all the same. + return true; + } + catch (AmazonS3Exception ex) when (IsConditionalWriteConflict(ex)) + { + _metrics.IncrementWriteConflicts(); + _logger.LogInformation( + "Public manifest {Key} changed concurrently during delete (attempt {Attempt}/{Max}); re-listing and retrying", + scope.RegistryKey, attempt, MaxWriteAttempts); + return false; + } + } + + private async Task TryPutManifest(ChangelogScope scope, string json, string? etag, int attempt, Cancel ctx) + { + var request = new PutObjectRequest + { + BucketName = publicBucketName, + Key = scope.RegistryKey, + ContentBody = json, + ContentType = "application/json" + }; + + // Optimistic concurrency: update only if unchanged, create only if still absent. + if (etag is null) + request.IfNoneMatch = "*"; + else + request.IfMatch = etag; + + try + { + _ = await s3Client.PutObjectAsync(request, ctx); + return true; + } + catch (AmazonS3Exception ex) when (IsConditionalWriteConflict(ex)) + { + _metrics.IncrementWriteConflicts(); + _logger.LogInformation( + "Public manifest {Key} changed concurrently (attempt {Attempt}/{Max}); re-listing and retrying", + scope.RegistryKey, attempt, MaxWriteAttempts); + return false; + } + } + + // 412 = a plain conditional-request loss; 409 = ConditionalRequestConflict, S3's signal for + // concurrent conditional writers on the same key. Both mean: re-read state and retry. + private static bool IsConditionalWriteConflict(AmazonS3Exception ex) => + ex.StatusCode is HttpStatusCode.PreconditionFailed or HttpStatusCode.Conflict; + + private async Task BackOff(int attempt, Cancel ctx) + { + if (_retryBaseDelay <= TimeSpan.Zero) + return; + var jitter = TimeSpan.FromMilliseconds(Random.Shared.NextDouble() * _retryBaseDelay.TotalMilliseconds); + await Task.Delay((_retryBaseDelay * attempt) + jitter, ctx); + } + + private static List Sort(IEnumerable entries) => + [.. entries + .OrderByDescending(b => VersionOrDate.Parse(b.Target ?? string.Empty)) + .ThenBy(b => b.File, StringComparer.Ordinal)]; + + private static string NormalizeETag(string? etag) => etag?.Trim('"') ?? string.Empty; + + private static bool BundlesEqual(IReadOnlyList a, IReadOnlyList b) + { + if (a.Count != b.Count) + return false; + + for (var i = 0; i < a.Count; i++) + { + if (!string.Equals(a[i].File, b[i].File, StringComparison.Ordinal) || + !string.Equals(a[i].Target, b[i].Target, StringComparison.Ordinal) || + !string.Equals(a[i].ETag, b[i].ETag, StringComparison.Ordinal)) + return false; + } + + return true; + } +} diff --git a/src/services/Elastic.Changelog/Scrubbing/ChangelogContentScrubber.cs b/src/services/Elastic.Changelog/Scrubbing/ChangelogContentScrubber.cs new file mode 100644 index 0000000000..9fd627e515 --- /dev/null +++ b/src/services/Elastic.Changelog/Scrubbing/ChangelogContentScrubber.cs @@ -0,0 +1,117 @@ +// 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.Changelog.Bundling; +using Elastic.Documentation.Configuration.ReleaseNotes; +using Elastic.Documentation.Diagnostics; +using Elastic.Documentation.ReleaseNotes; +using Microsoft.Extensions.Logging; + +namespace Elastic.Changelog.Scrubbing; + +/// Rewrites private-bucket changelog YAML into its public, allowlist-scrubbed form. +public interface IChangelogContentScrubber +{ + /// + /// Scrubs for public publication. The key decides the document + /// shape: bundle/{product}/… is a bundle, everything else a changelog entry. Throws + /// when the content cannot be proven free of private references. + /// + Task ScrubAsync(string key, string content, Cancel ctx); +} + +/// +/// The scrub pass previously inlined in the scrubber Lambda's Program.cs: applies the +/// repository allowlist via and validates the result before +/// it may reach the public bucket. +/// +public sealed class ChangelogContentScrubber(ILoggerFactory logFactory, IReadOnlyList allowRepos) : IChangelogContentScrubber +{ + private readonly ILogger _logger = logFactory.CreateLogger(); + + /// + public async Task ScrubAsync(string key, string content, Cancel ctx) + { + // Artifact-root layout: bundles live under "bundle/{product}/…", entries under + // "changelog/{org}/{repo}/{branch}/…". Match the bundle prefix (not a "/bundle/" substring, + // which no longer appears in the new keys) so bundles are not misclassified as entries. + var isBundlePath = key.StartsWith(ChangelogKeys.BundlePrefix, StringComparison.OrdinalIgnoreCase); + + return isBundlePath + ? await ScrubBundle(content, ctx) + : await ScrubChangelog(content, ctx); + } + + private async Task ScrubBundle(string content, Cancel ctx) + { + ctx.ThrowIfCancellationRequested(); + + var bundle = ReleaseNotesSerialization.DeserializeBundle(content); + var owner = bundle.Products.Count > 0 ? bundle.Products[0].Owner ?? "elastic" : "elastic"; + var repo = bundle.Products.Count > 0 ? bundle.Products[0].Repo : null; + + await using var collector = new DiagnosticsCollector([]); + if (!LinkAllowlistSanitizer.ScrubBundleForPublic(collector, bundle, allowRepos, owner, repo, out var sanitized, out var changed)) + throw new InvalidOperationException($"Failed to scrub bundle for public output; errors: {collector.Errors}"); + + if (!changed) + { + _logger.LogInformation("Bundle had no private references, writing unchanged"); + LinkAllowlistSanitizer.ValidateNoPrivateReferences(content, allowRepos); + return content; + } + + var result = ReleaseNotesSerialization.SerializeBundle(sanitized); + LinkAllowlistSanitizer.ValidateNoPrivateReferences(result, allowRepos); + return result; + } + + private async Task ScrubChangelog(string content, Cancel ctx) + { + ctx.ThrowIfCancellationRequested(); + + var normalized = ReleaseNotesSerialization.NormalizeYaml(content); + var entry = ReleaseNotesSerialization.DeserializeEntry(normalized); + + var bundledEntry = new BundledEntry + { + Type = entry.Type, + Title = entry.Title, + Description = entry.Description, + Impact = entry.Impact, + Action = entry.Action, + Prs = entry.Prs, + Issues = entry.Issues, + Areas = entry.Areas, + Highlight = entry.Highlight, + Subtype = entry.Subtype + }; + + await using var collector = new DiagnosticsCollector([]); + if (!LinkAllowlistSanitizer.TryApplyChangelogEntry( + collector, bundledEntry, allowRepos, "elastic", null, + out var sanitized, out var changed)) + throw new InvalidOperationException($"Failed to apply allowlist to changelog entry; errors: {collector.Errors}"); + + if (!changed) + { + _logger.LogInformation("Changelog entry had no private references, writing unchanged"); + LinkAllowlistSanitizer.ValidateNoPrivateReferences(content, allowRepos); + return content; + } + + var scrubEntry = entry with + { + Description = sanitized.Description, + Impact = sanitized.Impact, + Action = sanitized.Action, + Prs = sanitized.Prs, + Issues = sanitized.Issues + }; + + var result = ReleaseNotesSerialization.SerializeEntry(scrubEntry); + LinkAllowlistSanitizer.ValidateNoPrivateReferences(result, allowRepos); + return result; + } +} diff --git a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs new file mode 100644 index 0000000000..a70af45507 --- /dev/null +++ b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs @@ -0,0 +1,285 @@ +// 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 System.Net; +using Amazon.S3; +using Amazon.S3.Model; +using Amazon.S3.Util; +using Elastic.Changelog.Reconciliation; +using Elastic.Documentation.Configuration.ReleaseNotes; +using Microsoft.Extensions.Logging; + +namespace Elastic.Changelog.Scrubbing; + +/// One SQS message as seen by the scrubber: its receipt identity and raw body. +public sealed record ScrubberQueueMessage(string MessageId, string Body); + +/// +/// The scrubber Lambda's event processor (elastic/docs-eng-team#688), extracted from +/// Program.cs so it is testable. Events are triggers, state decides: the handler never +/// acts on an event's type — an event means only "this key may have changed, look at +/// it". Every distinct key gets one object-level reconcile against the private bucket, then every +/// distinct group gets one registry reconcile against the public listing, so out-of-order and +/// at-least-once S3 notifications are harmless and each batch heals accumulated drift. +/// +public sealed class ScrubberProcessor( + ILoggerFactory logFactory, + IAmazonS3 s3Client, + string publicBucketName, + IChangelogContentScrubber scrubber, + RegistryReconciler reconciler, + ReconcileMetrics? metrics = null +) +{ + // Bounds the reread-and-redo loop of post-write source validation. Each redo only triggers + // when the private object changed mid-flight, which itself queued another event; converging + // here is an optimization, not a correctness requirement. + private const int MaxObjectAttempts = 3; + + private readonly ILogger _logger = logFactory.CreateLogger(); + private readonly ReconcileMetrics _metrics = metrics ?? new ReconcileMetrics(); + + private sealed class ObjectWork(string sourceBucket) + { + public string SourceBucket { get; set; } = sourceBucket; + public HashSet MessageIds { get; } = [with(StringComparer.Ordinal)]; + } + + private sealed class GroupWork(ChangelogScope scope) + { + public ChangelogScope Scope { get; } = scope; + public HashSet MessageIds { get; } = [with(StringComparer.Ordinal)]; + } + + /// + /// Processes one SQS batch and returns the message ids that must be redelivered. Work is + /// coalesced per distinct key and per distinct group; a failed object reconcile fails every + /// message that referenced its key, a failed group reconcile fails every message that + /// contributed to that group. + /// + public async Task> ProcessAsync(IReadOnlyList messages, Cancel ctx) + { + var objectWork = new Dictionary(StringComparer.Ordinal); + var groupWork = new Dictionary(StringComparer.Ordinal); + var failedIds = new HashSet(StringComparer.Ordinal); + + foreach (var message in messages) + { + try + { + var s3Event = S3EventNotification.ParseJson(message.Body); + foreach (var record in s3Event.Records ?? []) + { + var key = Uri.UnescapeDataString(record.S3.Object.Key.Replace('+', ' ')); + _logger.LogInformation("Batch names key={Key} (event={EventName})", key, record.EventName?.Value); + Classify(message.MessageId, record.S3.Bucket.Name, key, objectWork, groupWork); + } + } + catch (Exception e) when (e is not OperationCanceledException) + { + _logger.LogWarning(e, "Failed to parse message {MessageId}", message.MessageId); + _ = failedIds.Add(message.MessageId); + } + } + + foreach (var (key, work) in objectWork) + { + ctx.ThrowIfCancellationRequested(); + try + { + await ReconcileObjectAsync(work.SourceBucket, key, ctx); + } + catch (Exception e) when (e is not OperationCanceledException) + { + _logger.LogError(e, "Object reconcile for {Key} failed; failing its {Count} message(s)", key, work.MessageIds.Count); + failedIds.UnionWith(work.MessageIds); + } + } + + foreach (var work in groupWork.Values) + { + ctx.ThrowIfCancellationRequested(); + try + { + _ = await reconciler.ReconcileGroupAsync(work.Scope, ctx); + } + catch (Exception e) when (e is not OperationCanceledException) + { + _logger.LogError(e, "Group reconcile for {Scope} failed; failing its {Count} contributing message(s)", work.Scope, work.MessageIds.Count); + failedIds.UnionWith(work.MessageIds); + } + } + + _metrics.AddFailedMessages(failedIds.Count); + return [.. failedIds]; + } + + private void Classify( + string messageId, + string sourceBucket, + string key, + Dictionary objectWork, + Dictionary groupWork) + { + var hasScope = ChangelogScope.TryFromKey(key, out var scope); + + // The registry pass-through is retired: registry keys are never copied or deleted. Old CLI + // versions still write private manifests (and Phase 3's cleanup will delete them) — those + // events only schedule a reconcile of the group, which derives the public manifest from + // public state. + if (ChangelogKeys.IsRegistry(key)) + { + if (hasScope) + AddGroup(groupWork, scope!, messageId); + return; + } + + if (key.EndsWith(".json", StringComparison.OrdinalIgnoreCase)) + { + _logger.LogWarning("Skipping unapproved JSON key: {Key}", key); + return; + } + + if (!key.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase) && + !key.EndsWith(".yml", StringComparison.OrdinalIgnoreCase)) + { + _logger.LogInformation("Skipping non-YAML key: {Key}", key); + return; + } + + if (!objectWork.TryGetValue(key, out var work)) + { + work = new ObjectWork(sourceBucket); + objectWork[key] = work; + } + work.SourceBucket = sourceBucket; + _ = work.MessageIds.Add(messageId); + + if (hasScope) + AddGroup(groupWork, scope!, messageId); + } + + private static void AddGroup(Dictionary groupWork, ChangelogScope scope, string messageId) + { + if (!groupWork.TryGetValue(scope.Prefix, out var work)) + { + work = new GroupWork(scope); + groupWork[scope.Prefix] = work; + } + _ = work.MessageIds.Add(messageId); + } + + /// + /// Order-independent object reconcile: the event type is ignored; the private bucket's current + /// state decides between scrub-and-copy and delete. A stale ObjectRemoved arriving + /// after a recreate re-copies the live object instead of deleting it. + /// + private async Task ReconcileObjectAsync(string sourceBucket, string key, Cancel ctx) + { + _metrics.IncrementObjectReconciles(); + + for (var attempt = 1; attempt <= MaxObjectAttempts; attempt++) + { + ctx.ThrowIfCancellationRequested(); + + var snapshot = await TryGetPrivateObject(sourceBucket, key, ctx); + if (snapshot is { } source) + { + var scrubbed = await scrubber.ScrubAsync(key, source.Content, ctx); + await PutPublicObject(key, scrubbed, ctx); + _logger.LogInformation("Scrubbed and wrote {Key} to public bucket", key); + } + else + { + await DeletePublicObject(key, ctx); + _logger.LogInformation("Private {Key} is gone; removed its public copy", key); + } + + // Post-write source validation: sequential out-of-order events are handled by the + // state read above, but two concurrent invocations can interleave so the older read + // publishes last. Confirm the private object still matches the snapshot this write + // was derived from; any change landing after this check has its own S3 event, so the + // combination converges. + var currentETag = await TryHeadPrivateObject(sourceBucket, key, ctx); + var stillCurrent = snapshot is null + ? currentETag is null + : string.Equals(currentETag, snapshot.Value.ETag, StringComparison.Ordinal); + if (stillCurrent) + return; + + _metrics.IncrementObjectReconcileRetries(); + _logger.LogInformation( + "Private {Key} changed while its reconcile was in flight (attempt {Attempt}/{Max}); redoing from current state", + key, attempt, MaxObjectAttempts); + } + + throw new InvalidOperationException( + $"Private {key} kept changing during {MaxObjectAttempts} reconcile attempts; failing the message for redelivery."); + } + + private async Task<(string Content, string ETag)?> TryGetPrivateObject(string sourceBucket, string key, Cancel ctx) + { + try + { + using var response = await s3Client.GetObjectAsync(new GetObjectRequest + { + BucketName = sourceBucket, + Key = key + }, ctx); + + await using var stream = response.ResponseStream; + using var reader = new StreamReader(stream); + var content = await reader.ReadToEndAsync(ctx); + return (content, NormalizeETag(response.ETag)); + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return null; + } + } + + private async Task TryHeadPrivateObject(string sourceBucket, string key, Cancel ctx) + { + try + { + var response = await s3Client.GetObjectMetadataAsync(new GetObjectMetadataRequest + { + BucketName = sourceBucket, + Key = key + }, ctx); + return NormalizeETag(response.ETag); + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return null; + } + } + + private async Task PutPublicObject(string key, string content, Cancel ctx) => + _ = await s3Client.PutObjectAsync(new PutObjectRequest + { + BucketName = publicBucketName, + Key = key, + ContentBody = content, + ContentType = "application/yaml" + }, ctx); + + private async Task DeletePublicObject(string key, Cancel ctx) + { + try + { + _ = await s3Client.DeleteObjectAsync(new DeleteObjectRequest + { + BucketName = publicBucketName, + Key = key + }, ctx); + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + // Already absent; converged. + } + } + + private static string NormalizeETag(string? etag) => etag?.Trim('"') ?? string.Empty; +} diff --git a/src/services/Elastic.Changelog/Uploading/Registry.cs b/src/services/Elastic.Changelog/Uploading/Registry.cs index 376810e1d7..ff81cb0b4c 100644 --- a/src/services/Elastic.Changelog/Uploading/Registry.cs +++ b/src/services/Elastic.Changelog/Uploading/Registry.cs @@ -18,10 +18,21 @@ namespace Elastic.Changelog.Uploading; /// public sealed record Registry { + /// The schema version written by this producer; consumers refuse newer versions. + public const int CurrentSchemaVersion = 1; + /// /// Manifest schema version. Incremented when consumers must change their parser. /// - public int SchemaVersion { get; init; } = 1; + public int SchemaVersion { get; init; } = CurrentSchemaVersion; + + /// + /// Identifies the algorithm (and its version) that produced this manifest. The registry + /// reconciler recomputes every entry — and rewrites the manifest even when the entries are + /// unchanged — whenever this differs from its own value, so metadata-logic changes roll out + /// deterministically. Null on manifests written by the legacy client-side refresh. + /// + public string? Producer { get; init; } /// /// Grouping identifier: the product for a bundle index (bundle/{product}/…) or the diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/FakeS3.cs b/tests/Elastic.Changelog.Tests/Reconciliation/FakeS3.cs new file mode 100644 index 0000000000..5681b2a982 --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Reconciliation/FakeS3.cs @@ -0,0 +1,231 @@ +// 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 System.Diagnostics.CodeAnalysis; +using System.Net; +using System.Security.Cryptography; +using System.Text; +using Amazon.S3; +using Amazon.S3.Model; +using FakeItEasy; + +namespace Elastic.Changelog.Tests.Reconciliation; + +/// +/// A stateful in-memory S3 behind a FakeItEasy , holding one or more +/// buckets so a single client can serve the scrubber's private-read/public-write flow. ETags are +/// the MD5 of the content (matching real single-part uploads), conditional PUTs and DELETEs +/// (If-Match / If-None-Match) enforce real 412 semantics, listings honor the +/// / delimiter and paginate with . Every read and write call is +/// recorded so tests can assert exactly what happened — or that nothing did. The hooks simulate +/// concurrent writers at precise interleaving points. +/// +internal sealed class FakeS3 +{ + private readonly Dictionary> _buckets = + [with(StringComparer.Ordinal)]; + + public IAmazonS3 Client { get; } = A.Fake(); + + /// Every PutObject call received, in order. + public List Puts { get; } = []; + + /// Every DeleteObject call received, in order. + public List Deletes { get; } = []; + + /// Every GetObject call received, in order. + public List Gets { get; } = []; + + /// Number of ListObjectsV2 calls received. + public int ListCalls { get; private set; } + + /// Objects per listing page, to exercise pagination. + public int PageSize { get; set; } = 1000; + + /// Runs before each PutObject is evaluated, with the 1-based call number — simulates a concurrent writer. + public Action? BeforePut { get; set; } + + /// Runs before each DeleteObject is evaluated, with the 1-based call number. + public Action? BeforeDelete { get; set; } + + /// Runs after a GetObject resolved its content (which is returned unchanged), with the key and 1-based call number — simulates the source changing right after a read. + public Action? AfterGet { get; set; } + + /// Runs before every ListObjectsV2 evaluation with the 1-based call number. + public Action? OnList { get; set; } + + private int _puts; + private int _deletes; + private int _gets; + + public FakeS3(params string[] bucketNames) + { + foreach (var bucket in bucketNames) + _buckets[bucket] = [with(StringComparer.Ordinal)]; + + _ = A.CallTo(() => Client.ListObjectsV2Async(A._, A._)) + .ReturnsLazily((ListObjectsV2Request r, CancellationToken _) => List(r)); + + _ = A.CallTo(() => Client.GetObjectAsync(A._, A._)) + .ReturnsLazily((GetObjectRequest r, CancellationToken _) => Get(r)); + + _ = A.CallTo(() => Client.GetObjectMetadataAsync(A._, A._)) + .ReturnsLazily((GetObjectMetadataRequest r, CancellationToken _) => Head(r)); + + _ = A.CallTo(() => Client.PutObjectAsync(A._, A._)) + .ReturnsLazily((PutObjectRequest r, CancellationToken _) => Put(r)); + + _ = A.CallTo(() => Client.DeleteObjectAsync(A._, A._)) + .ReturnsLazily((DeleteObjectRequest r, CancellationToken _) => Delete(r)); + } + + // MD5 is what real S3 uses for single-part ETags. + [SuppressMessage("Security", "CA5351:Do Not Use Broken Cryptographic Algorithms")] + public static string ETagOf(string content) => + Convert.ToHexStringLower(MD5.HashData(Encoding.UTF8.GetBytes(content))); + + /// Seeds or replaces an object; returns its (unquoted) ETag. + public string Seed(string bucket, string key, string content) + { + var etag = ETagOf(content); + Store(bucket)[key] = (content, etag); + return etag; + } + + public void Remove(string bucket, string key) => Store(bucket).Remove(key); + + public bool Exists(string bucket, string key) => Store(bucket).ContainsKey(key); + + public string ContentOf(string bucket, string key) => Store(bucket)[key].Content; + + /// The keys of every GetObject call for . + public IReadOnlyList GetsFor(string bucket) => + [.. Gets.Where(g => string.Equals(g.BucketName, bucket, StringComparison.Ordinal)).Select(g => g.Key)]; + + private Dictionary Store(string bucket) => + _buckets.TryGetValue(bucket, out var store) + ? store + : throw new InvalidOperationException($"Bucket {bucket} was not declared on this FakeS3"); + + private ListObjectsV2Response List(ListObjectsV2Request request) + { + ListCalls++; + OnList?.Invoke(ListCalls); + + var store = Store(request.BucketName); + var prefix = request.Prefix ?? string.Empty; + var objects = new List(); + var commonPrefixes = new SortedSet(StringComparer.Ordinal); + + foreach (var (key, value) in store.Where(kv => kv.Key.StartsWith(prefix, StringComparison.Ordinal)).OrderBy(kv => kv.Key, StringComparer.Ordinal)) + { + var rest = key[prefix.Length..]; + var slash = rest.IndexOf('/', StringComparison.Ordinal); + if (request.Delimiter == "/" && slash >= 0) + { + _ = commonPrefixes.Add(prefix + rest[..(slash + 1)]); + continue; + } + + objects.Add(new S3Object + { + Key = key, + ETag = $"\"{value.ETag}\"", + Size = value.Content.Length, + LastModified = new DateTime(2026, 5, 6, 12, 0, 0, DateTimeKind.Utc) + }); + } + + var start = request.ContinuationToken is { } token ? int.Parse(token, System.Globalization.CultureInfo.InvariantCulture) : 0; + var page = objects.Skip(start).Take(PageSize).ToList(); + var truncated = start + page.Count < objects.Count; + + return new ListObjectsV2Response + { + S3Objects = page, + CommonPrefixes = [.. commonPrefixes], + IsTruncated = truncated, + NextContinuationToken = truncated + ? (start + page.Count).ToString(System.Globalization.CultureInfo.InvariantCulture) + : null + }; + } + + private GetObjectResponse Get(GetObjectRequest request) + { + Gets.Add(request); + _gets++; + + if (!Store(request.BucketName).TryGetValue(request.Key, out var obj)) + throw NotFound(); + + // Capture the response before the hook runs, so a hook that reseeds the key simulates a + // write landing right after this read. + var response = new GetObjectResponse + { + ETag = $"\"{obj.ETag}\"", + ResponseStream = new MemoryStream(Encoding.UTF8.GetBytes(obj.Content)) + }; + AfterGet?.Invoke(request.Key, _gets); + return response; + } + + private GetObjectMetadataResponse Head(GetObjectMetadataRequest request) + { + if (!Store(request.BucketName).TryGetValue(request.Key, out var obj)) + throw NotFound(); + + var response = new GetObjectMetadataResponse + { + ETag = $"\"{obj.ETag}\"" + }; + response.Headers.ContentType = "application/yaml"; + return response; + } + + private PutObjectResponse Put(PutObjectRequest request) + { + _puts++; + BeforePut?.Invoke(_puts); + + Puts.Add(request); + + var store = Store(request.BucketName); + var exists = store.TryGetValue(request.Key, out var current); + if (request.IfNoneMatch == "*" && exists) + throw PreconditionFailed(); + if (request.IfMatch is { } ifMatch && (!exists || ifMatch.Trim('"') != current.ETag)) + throw PreconditionFailed(); + + _ = Seed(request.BucketName, request.Key, request.ContentBody); + return new PutObjectResponse(); + } + + private DeleteObjectResponse Delete(DeleteObjectRequest request) + { + _deletes++; + BeforeDelete?.Invoke(_deletes); + + Deletes.Add(request); + + var store = Store(request.BucketName); + var exists = store.TryGetValue(request.Key, out var current); + if (request.IfMatch is { } ifMatch) + { + if (!exists) + throw NotFound(); + if (ifMatch.Trim('"') != current.ETag) + throw PreconditionFailed(); + } + + _ = store.Remove(request.Key); + return new DeleteObjectResponse(); + } + + private static AmazonS3Exception NotFound() => + new("Not Found") { StatusCode = HttpStatusCode.NotFound }; + + private static AmazonS3Exception PreconditionFailed() => + new("Precondition Failed") { StatusCode = HttpStatusCode.PreconditionFailed }; +} diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs new file mode 100644 index 0000000000..607212e2d2 --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs @@ -0,0 +1,463 @@ +// 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 System.Text.Json; +using AwesomeAssertions; +using Elastic.Changelog.Reconciliation; +using Elastic.Changelog.Uploading; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Elastic.Changelog.Tests.Reconciliation; + +public class RegistryReconcilerTests +{ + private const string PublicBucket = "public-bucket"; + + private static readonly DateTimeOffset FixedNow = new(2026, 7, 28, 12, 0, 0, TimeSpan.Zero); + + private readonly FakeS3 _s3 = new(PublicBucket); + private readonly RegistryReconciler _reconciler; + private readonly ReconcileMetrics _metrics = new(); + + public RegistryReconcilerTests() => + _reconciler = new RegistryReconciler( + NullLoggerFactory.Instance, + _s3.Client, + PublicBucket, + new FakeTimeProvider(FixedNow), + retryBaseDelay: TimeSpan.Zero, + _metrics); + + private static ChangelogScope BundleScope(string product = "elasticsearch") + { + _ = ChangelogScope.TryCreateBundle(product, out var scope); + return scope!; + } + + private static ChangelogScope ChangelogScopeFor(string org, string repo, string branch) + { + _ = ChangelogScope.TryCreateChangelog(org, repo, branch, out var scope); + return scope!; + } + + // language=yaml + private static string BundleYaml(string product, string target) => $""" + products: + - product: {product} + target: {target} + repo: {product} + owner: elastic + entries: + - file: + name: 1-feature.yaml + checksum: deadbeef + type: enhancement + title: Sample + """; + + // language=yaml + private const string AmendYaml = """ + exclude-entries: + - file: + name: 1-feature.yaml + checksum: deadbeef + """; + + private string SeedBundle(ChangelogScope scope, string file, string target, string? product = null) => + _s3.Seed(PublicBucket, scope.Prefix + file, BundleYaml(product ?? scope.Group, target)); + + private void SeedManifest(ChangelogScope scope, params RegistryBundle[] bundles) => + SeedManifest(scope, RegistryReconciler.Producer, Registry.CurrentSchemaVersion, bundles); + + private void SeedManifest(ChangelogScope scope, string? producer, int schemaVersion, params RegistryBundle[] bundles) + { + var manifest = new Registry + { + SchemaVersion = schemaVersion, + Product = scope.Group, + Producer = producer, + GeneratedAt = FixedNow.AddDays(-1), + Bundles = bundles + }; + _ = _s3.Seed(PublicBucket, scope.RegistryKey, JsonSerializer.Serialize(manifest, RegistryJsonContext.Default.Registry)); + } + + private Registry WrittenManifest() + { + var content = _s3.ContentOf(PublicBucket, BundleScope().RegistryKey); + return JsonSerializer.Deserialize(content, RegistryJsonContext.Default.Registry)!; + } + + private Cancel Ctx => TestContext.Current.CancellationToken; + + [Fact] + public async Task ReconcileGroup_HealsEntriesMissingFromManifest() + { + // Public bucket holds 1..4 but the manifest lists only 1 and 2 (3 was lost to a past gap; + // 4 just landed): a single reconcile heals both, and only reads the YAMLs it cannot reuse. + var scope = BundleScope(); + var etag1 = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + var etag2 = SeedBundle(scope, "es-9.2.0.yaml", "9.2.0"); + _ = SeedBundle(scope, "es-9.3.0.yaml", "9.3.0"); + _ = SeedBundle(scope, "es-9.4.0.yaml", "9.4.0"); + SeedManifest(scope, + new RegistryBundle { File = "es-9.2.0.yaml", Target = "9.2.0", ETag = etag2 }, + new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag1 }); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var manifest = WrittenManifest(); + manifest.Bundles.Select(b => b.File).Should().Equal( + "es-9.4.0.yaml", "es-9.3.0.yaml", "es-9.2.0.yaml", "es-9.1.0.yaml"); + manifest.Bundles.Should().OnlyContain(b => b.Target != null); + manifest.Producer.Should().Be(RegistryReconciler.Producer); + + // 1 and 2 were ETag-reused: only the manifest itself plus 3 and 4 were read. + _s3.GetsFor(PublicBucket).Should().BeEquivalentTo([ + scope.RegistryKey, scope.Prefix + "es-9.3.0.yaml", scope.Prefix + "es-9.4.0.yaml" + ]); + } + + [Fact] + public async Task ReconcileGroup_DropsEntriesWhoseObjectIsGone() + { + var scope = BundleScope(); + var etag1 = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope, + new RegistryBundle { File = "es-9.2.0.yaml", Target = "9.2.0", ETag = "gone" }, + new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag1 }); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + WrittenManifest().Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); + } + + [Fact] + public async Task ReconcileGroup_ManifestAlreadyExact_SkipsWriteAndBundleReads() + { + var scope = BundleScope(); + var etag1 = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + var etag2 = SeedBundle(scope, "es-9.2.0.yaml", "9.2.0"); + SeedManifest(scope, + new RegistryBundle { File = "es-9.2.0.yaml", Target = "9.2.0", ETag = etag2 }, + new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag1 }); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Unchanged); + _s3.Puts.Should().BeEmpty(); + _s3.GetsFor(PublicBucket).Should().Equal(scope.RegistryKey); + _metrics.RegistryUnchanged.Should().Be(1); + } + + [Fact] + public async Task ReconcileGroup_GeneratedAtAloneNeverCausesChurn() + { + // Same as above but the seeded generated_at differs from "now": still Unchanged. + var scope = BundleScope(); + var etag = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope, new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Unchanged); + _s3.Puts.Should().BeEmpty(); + } + + [Fact] + public async Task ReconcileGroup_AmendIsAlwaysRecomputed_EvenWhenItsETagMatches() + { + // The parent moved from 9.3.0 to 9.4.0 without the amend's own ETag changing; an + // ETag-reuse of the amend entry would keep the stale inherited target forever. + var scope = BundleScope(); + var parentETag = SeedBundle(scope, "es-9.3.0.yaml", "9.4.0"); + var amendETag = _s3.Seed(PublicBucket, scope.Prefix + "es-9.3.0.amend-1.yaml", AmendYaml); + SeedManifest(scope, + new RegistryBundle { File = "es-9.3.0.yaml", Target = "9.4.0", ETag = parentETag }, + new RegistryBundle { File = "es-9.3.0.amend-1.yaml", Target = "9.3.0", ETag = amendETag }); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var amend = WrittenManifest().Bundles.Single(b => b.File == "es-9.3.0.amend-1.yaml"); + amend.Target.Should().Be("9.4.0", "the amend re-inherits the parent's current target on every reconcile"); + } + + [Fact] + public async Task ReconcileGroup_AmendWithoutParent_RecordsNullTarget() + { + var scope = BundleScope(); + _ = _s3.Seed(PublicBucket, scope.Prefix + "es-9.3.0.amend-1.yaml", AmendYaml); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var entry = WrittenManifest().Bundles.Should().ContainSingle().Subject; + entry.File.Should().Be("es-9.3.0.amend-1.yaml"); + entry.Target.Should().BeNull("the parent has not landed yet; a later reconcile self-corrects"); + } + + [Fact] + public async Task ReconcileGroup_MultiProductBundle_MatchesTheGroupProduct() + { + // language=yaml + const string multiProductYaml = """ + products: + - product: elasticsearch + target: 9.3.0 + repo: elasticsearch + owner: elastic + - product: kibana + target: 9.4.0 + repo: kibana + owner: elastic + entries: + - file: + name: 1-feature.yaml + checksum: deadbeef + type: enhancement + title: Sample + """; + var scope = BundleScope("kibana"); + _ = _s3.Seed(PublicBucket, scope.Prefix + "multi.yaml", multiProductYaml); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var content = _s3.ContentOf(PublicBucket, scope.RegistryKey); + var manifest = JsonSerializer.Deserialize(content, RegistryJsonContext.Default.Registry)!; + manifest.Bundles.Single().Target.Should().Be("9.4.0", "kibana's own target must win, never blindly Products[0]"); + } + + [Fact] + public async Task ReconcileGroup_ProducerMismatch_RecomputesEverythingAndWritesEvenWhenIdentical() + { + // A legacy (pass-through) manifest has no producer. Even when every entry would come out + // identical, the write must happen — otherwise the producer version is never adopted and + // every future reconcile keeps recomputing. + var scope = BundleScope(); + var etag = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope, producer: null, Registry.CurrentSchemaVersion, + new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + _s3.GetsFor(PublicBucket).Should().Contain(scope.Prefix + "es-9.1.0.yaml", "producer mismatch disables ETag reuse"); + var manifest = WrittenManifest(); + manifest.Producer.Should().Be(RegistryReconciler.Producer); + manifest.Bundles.Single().Should().BeEquivalentTo(new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); + } + + [Fact] + public async Task ReconcileGroup_ListingPaginates() + { + var scope = BundleScope(); + for (var i = 1; i <= 5; i++) + _ = SeedBundle(scope, $"es-9.{i}.0.yaml", $"9.{i}.0"); + _s3.PageSize = 2; + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + WrittenManifest().Bundles.Should().HaveCount(5); + _s3.ListCalls.Should().BeGreaterThanOrEqualTo(3); + } + + [Fact] + public async Task ReconcileGroup_BranchWithSlash_DoesNotSweepNestedPools() + { + // The "main" pool must not swallow the "main/feature" pool: branches are stored verbatim, + // so the / delimiter is what keeps them apart. + var scope = ChangelogScopeFor("elastic", "repo", "main"); + _ = _s3.Seed(PublicBucket, "changelog/elastic/repo/main/entry-a.yaml", "a: 1"); + _ = _s3.Seed(PublicBucket, "changelog/elastic/repo/main/feature/entry-b.yaml", "b: 1"); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var content = _s3.ContentOf(PublicBucket, scope.RegistryKey); + var manifest = JsonSerializer.Deserialize(content, RegistryJsonContext.Default.Registry)!; + manifest.Bundles.Select(b => b.File).Should().Equal("entry-a.yaml"); + } + + [Fact] + public async Task ReconcileGroup_ChangelogScope_IsListingOnly_ZeroObjectReads() + { + var scope = ChangelogScopeFor("elastic", "repo", "main"); + _ = _s3.Seed(PublicBucket, scope.Prefix + "entry-a.yaml", "a: 1"); + _ = _s3.Seed(PublicBucket, scope.Prefix + "entry-b.yaml", "b: 1"); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var content = _s3.ContentOf(PublicBucket, scope.RegistryKey); + var manifest = JsonSerializer.Deserialize(content, RegistryJsonContext.Default.Registry)!; + manifest.Bundles.Should().HaveCount(2); + manifest.Bundles.Should().OnlyContain(b => b.Target == null, "the entry index is listing-only by design"); + _s3.GetsFor(PublicBucket).Should().Equal(scope.RegistryKey); + } + + [Fact] + public async Task ReconcileGroup_ExcludesTheManifestAndOtherNonYamlFromItsOwnListing() + { + var scope = BundleScope(); + _ = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope); + _ = _s3.Seed(PublicBucket, scope.Prefix + "notes.txt", "not yaml"); + _ = _s3.Seed(PublicBucket, scope.Prefix + "stray.json", "{}"); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + WrittenManifest().Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); + } + + [Fact] + public async Task ReconcileGroup_CorruptManifest_IsRebuiltWithItsLiveETagGuard() + { + var scope = BundleScope(); + _ = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + var corruptETag = _s3.Seed(PublicBucket, scope.RegistryKey, "{ not json "); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + var put = _s3.Puts.Should().ContainSingle().Subject; + put.IfMatch.Trim('"').Should().Be(corruptETag, "the conditional write must replace exactly the corrupt manifest that was read"); + WrittenManifest().Bundles.Should().ContainSingle(); + } + + [Fact] + public async Task ReconcileGroup_EmptyListing_DeletesTheManifestConditionally() + { + var scope = BundleScope(); + SeedManifest(scope, new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = "aa" }); + var manifestETag = FakeS3.ETagOf(_s3.ContentOf(PublicBucket, scope.RegistryKey)); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Deleted); + _s3.Exists(PublicBucket, scope.RegistryKey).Should().BeFalse(); + var delete = _s3.Deletes.Should().ContainSingle().Subject; + delete.IfMatch.Trim('"').Should().Be(manifestETag); + } + + [Fact] + public async Task ReconcileGroup_EmptyListing_DeletesEvenAManifestWhoseBundlesAreAlreadyEmpty() + { + // Deletion must run before any equality short-circuit: absent ≠ empty for consumers. + var scope = BundleScope(); + SeedManifest(scope); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Deleted); + _s3.Exists(PublicBucket, scope.RegistryKey).Should().BeFalse(); + } + + [Fact] + public async Task ReconcileGroup_EmptyListingAndNoManifest_IsANoOp() + { + var outcome = await _reconciler.ReconcileGroupAsync(BundleScope(), Ctx); + + outcome.Should().Be(GroupReconcileOutcome.NoOp); + _s3.Puts.Should().BeEmpty(); + _s3.Deletes.Should().BeEmpty(); + } + + [Fact] + public async Task ReconcileGroup_DeleteLosingTheRace_RereadsAndRetries() + { + // A concurrent reconciler replaces the manifest between our read and our delete: the + // conditional delete 412s, and the retry deletes the fresh manifest it re-reads. + var scope = BundleScope(); + SeedManifest(scope); + _s3.BeforeDelete = call => + { + if (call == 1) + SeedManifest(scope, new RegistryBundle { File = "late.yaml", Target = null, ETag = "bb" }); + }; + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Deleted); + _s3.Exists(PublicBucket, scope.RegistryKey).Should().BeFalse(); + _s3.Deletes.Should().HaveCount(2); + _metrics.WriteConflicts.Should().Be(1); + } + + [Fact] + public async Task ReconcileGroup_PutLosingTheRace_RereadsAndRetries() + { + var scope = BundleScope(); + _ = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope, producer: null, Registry.CurrentSchemaVersion); + _s3.BeforePut = call => + { + if (call == 1) + SeedManifest(scope, producer: null, Registry.CurrentSchemaVersion, + new RegistryBundle { File = "concurrent.yaml", Target = null, ETag = "cc" }); + }; + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.Written); + _s3.Puts.Should().HaveCount(2); + WrittenManifest().Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); + } + + [Fact] + public async Task ReconcileGroup_ExhaustedConditionalRetries_Throws() + { + var scope = BundleScope(); + _ = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope, producer: null, Registry.CurrentSchemaVersion); + // Every attempt loses: a concurrent writer lands between every read and write. + var counter = 0; + _s3.BeforePut = _ => SeedManifest(scope, producer: null, Registry.CurrentSchemaVersion, + new RegistryBundle { File = $"concurrent-{counter++}.yaml", Target = null, ETag = "cc" }); + + var act = async () => await _reconciler.ReconcileGroupAsync(scope, Ctx); + + _ = await act.Should().ThrowAsync(); + _s3.Puts.Should().HaveCount(5, "the retry loop is bounded"); + } + + [Fact] + public async Task ReconcileGroup_NewerSchemaManifest_IsReportedAndLeftUntouched() + { + var scope = BundleScope(); + _ = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); + SeedManifest(scope, RegistryReconciler.Producer, schemaVersion: Registry.CurrentSchemaVersion + 1); + var before = _s3.ContentOf(PublicBucket, scope.RegistryKey); + + var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + outcome.Should().Be(GroupReconcileOutcome.RefusedNewerSchema); + _s3.Puts.Should().BeEmpty(); + _s3.Deletes.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, scope.RegistryKey).Should().Be(before); + } + + [Fact] + public async Task ReconcileGroup_SortsNewestTargetFirstWithFileNameTiebreak() + { + var scope = BundleScope(); + _ = SeedBundle(scope, "b.yaml", "9.1.0"); + _ = SeedBundle(scope, "a.yaml", "9.1.0"); + _ = SeedBundle(scope, "c.yaml", "9.4.0"); + + _ = await _reconciler.ReconcileGroupAsync(scope, Ctx); + + WrittenManifest().Bundles.Select(b => b.File).Should().Equal("c.yaml", "a.yaml", "b.yaml"); + } + + private sealed class FakeTimeProvider(DateTimeOffset now) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => now; + } +} diff --git a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs new file mode 100644 index 0000000000..5fb95d40b6 --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs @@ -0,0 +1,286 @@ +// 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 System.Text.Json; +using AwesomeAssertions; +using Elastic.Changelog.Reconciliation; +using Elastic.Changelog.Scrubbing; +using Elastic.Changelog.Tests.Reconciliation; +using Elastic.Changelog.Uploading; +using FakeItEasy; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Elastic.Changelog.Tests.Scrubbing; + +public class ScrubberProcessorTests +{ + private const string PrivateBucket = "private-bucket"; + private const string PublicBucket = "public-bucket"; + + private readonly FakeS3 _s3 = new(PrivateBucket, PublicBucket); + private readonly IChangelogContentScrubber _scrubber = A.Fake(); + private readonly ReconcileMetrics _metrics = new(); + private readonly ScrubberProcessor _processor; + + public ScrubberProcessorTests() + { + // The real scrub pass has its own tests; here it just marks content so assertions can + // tell a scrubbed write from a raw copy. + _ = A.CallTo(() => _scrubber.ScrubAsync(A._, A._, A._)) + .ReturnsLazily((string _, string content, Cancel _) => Task.FromResult("scrubbed: " + content)); + + var reconciler = new RegistryReconciler( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); + _processor = new ScrubberProcessor( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, _metrics); + } + + private Cancel Ctx => TestContext.Current.CancellationToken; + + private static int MessageCounter; + + private static ScrubberQueueMessage Message(string eventName, string key, string bucket = PrivateBucket) + { + var id = $"msg-{Interlocked.Increment(ref MessageCounter)}"; + // The shape S3 bucket notifications deliver to SQS (fields the processor reads). + var body = + "{\"Records\":[{\"eventName\":\"" + eventName + "\",\"s3\":{\"bucket\":{\"name\":\"" + bucket + + "\"},\"object\":{\"key\":\"" + key + "\"}}}]}"; + return new ScrubberQueueMessage(id, body); + } + + private Registry PublicManifest(string registryKey) => + JsonSerializer.Deserialize(_s3.ContentOf(PublicBucket, registryKey), RegistryJsonContext.Default.Registry)!; + + [Fact] + public async Task Process_CreatedEvent_ScrubsCopiesAndWritesTheGroupManifest() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "content-1"); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml")], Ctx); + + failed.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml").Should().Be("scrubbed: content-1"); + PublicManifest("bundle/elasticsearch/registry.json").Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); + } + + [Fact] + public async Task Process_StaleRemovedEventAfterRecreate_RecopiesInsteadOfDeleting() + { + // The event type is advisory: the private object exists again, so a late ObjectRemoved + // must re-copy the live object, not delete the public one. + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "recreated"); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "old-public"); + + var failed = await _processor.ProcessAsync([Message("ObjectRemoved:Delete", "bundle/elasticsearch/es-9.1.0.yaml")], Ctx); + + failed.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml").Should().Be("scrubbed: recreated"); + _s3.Deletes.Should().NotContain(d => d.Key == "bundle/elasticsearch/es-9.1.0.yaml"); + } + + [Fact] + public async Task Process_StaleCreatedEventAfterDelete_RemovesThePublicCopyAndManifest() + { + // Private object is gone; a late ObjectCreated must converge on deletion — and the group + // reconcile then removes the now-empty group's manifest. + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "stale-public"); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/registry.json", + /*lang=json,strict*/ """{"schema_version":1,"product":"elasticsearch","generated_at":"2026-01-01T00:00:00+00:00","bundles":[]}"""); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml")], Ctx); + + failed.Should().BeEmpty(); + _s3.Exists(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml").Should().BeFalse(); + _s3.Exists(PublicBucket, "bundle/elasticsearch/registry.json").Should().BeFalse("an empty group's manifest is deleted: absent ≠ empty"); + } + + [Fact] + public async Task Process_RegistryKeyEvents_NeverCopyOrDelete_OnlyTriggerAGroupReconcile() + { + // Old CLI versions still write private manifests; Phase 3's cleanup will delete them. + // Neither event may touch the public registry object directly — but each schedules a + // reconcile that derives the public manifest from public state. + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/registry.json", /*lang=json,strict*/ """{"private":"manifest"}"""); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", BundleYaml()); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "bundle/elasticsearch/registry.json")], Ctx); + + failed.Should().BeEmpty(); + // The public manifest was reconciled from the listing — not copied from the private one. + var manifest = PublicManifest("bundle/elasticsearch/registry.json"); + manifest.Producer.Should().Be(RegistryReconciler.Producer); + manifest.Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); + _s3.GetsFor(PrivateBucket).Should().BeEmpty("the private registry content must never be read for pass-through"); + } + + [Fact] + public async Task Process_OtherJsonAndNonYamlKeys_AreSkipped() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/stray.json", "{}"); + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/notes.txt", "text"); + + var failed = await _processor.ProcessAsync( + [ + Message("ObjectCreated:Put", "bundle/elasticsearch/stray.json"), + Message("ObjectCreated:Put", "bundle/elasticsearch/notes.txt") + ], Ctx); + + failed.Should().BeEmpty(); + _s3.Puts.Should().BeEmpty(); + _s3.Deletes.Should().BeEmpty(); + } + + [Fact] + public async Task Process_MultipleEventsForOneKey_CoalesceIntoASingleObjectReconcile() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "content"); + + var failed = await _processor.ProcessAsync( + [ + Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml"), + Message("ObjectRemoved:Delete", "bundle/elasticsearch/es-9.1.0.yaml"), + Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml") + ], Ctx); + + failed.Should().BeEmpty(); + _metrics.ObjectReconciles.Should().Be(1, "the event type is ignored, so one key needs one look"); + _s3.Puts.Where(p => p.Key == "bundle/elasticsearch/es-9.1.0.yaml").Should().ContainSingle(); + } + + [Fact] + public async Task Process_MultipleKeysInOneGroup_CoalesceIntoASingleGroupReconcile() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.2.0.yaml", "two"); + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.3.0.yaml", "three"); + + var failed = await _processor.ProcessAsync( + [ + Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml"), + Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.2.0.yaml"), + Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.3.0.yaml") + ], Ctx); + + failed.Should().BeEmpty(); + _metrics.ObjectReconciles.Should().Be(3); + _metrics.GroupReconciles.Should().Be(1, "all three keys share one group"); + PublicManifest("bundle/elasticsearch/registry.json").Bundles.Should().HaveCount(3); + } + + [Fact] + public async Task Process_SourceChangingMidFlight_IsDetectedByPostWriteValidationAndRedone() + { + // Older-read-writes-last: v2 lands right after our read of v1. The post-write HEAD sees + // the mismatch and redoes the reconcile from current state, so v1 never wins. + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "v1"); + _s3.AfterGet = (key, call) => + { + if (key == "bundle/elasticsearch/es-9.1.0.yaml" && _s3.ContentOf(PrivateBucket, key) == "v1") + _ = _s3.Seed(PrivateBucket, key, "v2"); + }; + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml")], Ctx); + + failed.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml").Should().Be("scrubbed: v2"); + _metrics.ObjectReconcileRetries.Should().Be(1); + } + + [Fact] + public async Task Process_FailedObjectReconcile_FailsOnlyItsOwnMessages() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/bad.yaml", "bad"); + _ = _s3.Seed(PrivateBucket, "bundle/kibana/good.yaml", "good"); + _ = A.CallTo(() => _scrubber.ScrubAsync("bundle/elasticsearch/bad.yaml", A._, A._)) + .Throws(new InvalidOperationException("cannot scrub")); + + var badMessage = Message("ObjectCreated:Put", "bundle/elasticsearch/bad.yaml"); + var goodMessage = Message("ObjectCreated:Put", "bundle/kibana/good.yaml"); + var failed = await _processor.ProcessAsync([badMessage, goodMessage], Ctx); + + failed.Should().ContainSingle().Which.Should().Be(badMessage.MessageId); + _s3.Exists(PublicBucket, "bundle/kibana/good.yaml").Should().BeTrue(); + PublicManifest("bundle/kibana/registry.json").Bundles.Should().ContainSingle(); + } + + [Fact] + public async Task Process_FailedGroupReconcile_FailsEveryContributingMessage() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.2.0.yaml", "two"); + _ = _s3.Seed(PrivateBucket, "bundle/kibana/kb-9.1.0.yaml", "three"); + // Make every conditional manifest write for the elasticsearch group lose its race. + var counter = 0; + _s3.BeforePut = call => + { + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/registry.json", $"{{\"race\":{counter++}}}"); + }; + + var first = Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml"); + var second = Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.2.0.yaml"); + var other = Message("ObjectCreated:Put", "bundle/kibana/kb-9.1.0.yaml"); + + var failed = await _processor.ProcessAsync([first, second, other], Ctx); + + failed.Should().BeEquivalentTo([first.MessageId, second.MessageId], + "every record that contributed to the failed group must redeliver"); + } + + [Fact] + public async Task Process_UnparseableMessageBody_FailsThatMessage() + { + var garbage = new ScrubberQueueMessage("msg-garbage", "not json at all {"); + + var failed = await _processor.ProcessAsync([garbage], Ctx); + + failed.Should().ContainSingle().Which.Should().Be("msg-garbage"); + } + + [Fact] + public async Task Process_KeyOutsideAnyGroupLayout_IsCopiedButTriggersNoGroupReconcile() + { + // changelog/{org}/{file} has too few segments for a pool; the object itself still syncs. + _ = _s3.Seed(PrivateBucket, "changelog/elastic/stray.yaml", "stray"); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "changelog/elastic/stray.yaml")], Ctx); + + failed.Should().BeEmpty(); + _s3.Exists(PublicBucket, "changelog/elastic/stray.yaml").Should().BeTrue(); + _metrics.GroupReconciles.Should().Be(0); + } + + [Fact] + public async Task Process_BatchMixingObjectAndRegistryEvents_MarksGroupContributionsAcrossBoth() + { + // A YAML event and a (retired pass-through) registry event for the same group coalesce + // into one group reconcile fed by both messages. + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/registry.json", "{}"); + + var failed = await _processor.ProcessAsync( + [ + Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml"), + Message("ObjectCreated:Put", "bundle/elasticsearch/registry.json") + ], Ctx); + + failed.Should().BeEmpty(); + _metrics.GroupReconciles.Should().Be(1); + } + + // language=yaml + private static string BundleYaml() => """ + products: + - product: elasticsearch + target: 9.1.0 + repo: elasticsearch + owner: elastic + entries: + - file: + name: 1-feature.yaml + checksum: deadbeef + type: enhancement + title: Sample + """; +} From 7f4181c683e2d24ee6e4ab722f421b2100e62cd6 Mon Sep 17 00:00:00 2001 From: Felipe Cotti Date: Thu, 6 Aug 2026 12:13:03 -0300 Subject: [PATCH 2/3] Scrubber: shallow per-tree registry maps; reconcile bundle manifests only MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework after review: per-pool changelog/{org}/{repo}/{branch}/registry.json manifests are no longer reconciled — they stay client-authored and are mirrored verbatim (state-decided, so deletes propagate) until Phase 3 retires them. Group reconciliation is scoped to bundle/{product}/, the only tree consumers enumerate. The scrubber now also maintains shallow per-tree maps (bundle/registry.json and changelog/registry.json) mapping each folder to an opaque change token, so caching layers can skip unchanged folders with one small read. The token digests the folder's sorted file/ETag listing rather than reusing the last-touched file's ETag: deleting an older file must change the token. --- .../EmfMetricsEmitter.cs | 133 ++++--- .../docs-lambda-changelog-scrubber/Program.cs | 3 +- .../Reconciliation/ReconcileMetrics.cs | 10 + .../Reconciliation/RegistryReconciler.cs | 35 +- .../ShallowRegistryReconciler.cs | 341 ++++++++++++++++++ .../Scrubbing/ScrubberProcessor.cs | 117 ++++-- .../Reconciliation/RegistryReconcilerTests.cs | 32 +- .../ShallowRegistryReconcilerTests.cs | 216 +++++++++++ .../Scrubbing/ScrubberProcessorTests.cs | 100 ++++- 9 files changed, 860 insertions(+), 127 deletions(-) create mode 100644 src/services/Elastic.Changelog/Reconciliation/ShallowRegistryReconciler.cs create mode 100644 tests/Elastic.Changelog.Tests/Reconciliation/ShallowRegistryReconcilerTests.cs diff --git a/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs b/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs index ffb34436f7..7b98796bd5 100644 --- a/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs +++ b/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs @@ -2,8 +2,8 @@ // 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 System.Text; using System.Text.Json; +using System.Text.Json.Serialization; using Elastic.Changelog.Reconciliation; namespace Elastic.Documentation.Lambda.ChangelogScrubber; @@ -11,67 +11,102 @@ namespace Elastic.Documentation.Lambda.ChangelogScrubber; /// /// Emits the per-invocation reconcile counters as a CloudWatch Embedded Metric Format line /// (elastic/docs-eng-team#688 Phase 0 observability: these numbers gate any later SQS/Lambda -/// tuning). Written with directly — no serializer registration, so -/// nothing for AOT trimming to miss — and to stdout unwrapped, which is what the EMF parser -/// requires. +/// tuning). The counter names are static, so the payload is a fixed source-generated contract; +/// it is written to stdout unwrapped, which is what the EMF parser requires. /// internal static class EmfMetricsEmitter { private const string Namespace = "docs-changelog-scrubber"; - private static readonly (string Name, Func Value)[] Counters = + private static readonly IReadOnlyList MetricDefinitions = [ - ("ObjectReconciles", m => m.ObjectReconciles), - ("ObjectReconcileRetries", m => m.ObjectReconcileRetries), - ("GroupReconciles", m => m.GroupReconciles), - ("RegistryWrites", m => m.RegistryWrites), - ("RegistryDeletes", m => m.RegistryDeletes), - ("RegistryUnchanged", m => m.RegistryUnchanged), - ("WriteConflicts", m => m.WriteConflicts), - ("ObjectsListed", m => m.ObjectsListed), - ("EntriesRecomputed", m => m.EntriesRecomputed), - ("FailedMessages", m => m.FailedMessages) + new() { Name = "ObjectReconciles" }, + new() { Name = "ObjectReconcileRetries" }, + new() { Name = "GroupReconciles" }, + new() { Name = "RegistryWrites" }, + new() { Name = "RegistryDeletes" }, + new() { Name = "RegistryUnchanged" }, + new() { Name = "WriteConflicts" }, + new() { Name = "ObjectsListed" }, + new() { Name = "EntriesRecomputed" }, + new() { Name = "ShallowRegistryWrites" }, + new() { Name = "ShallowRegistryUnchanged" }, + new() { Name = "FailedMessages" } ]; public static void Emit(ReconcileMetrics metrics) { - using var buffer = new MemoryStream(); - using (var writer = new Utf8JsonWriter(buffer)) + var payload = new EmfPayload { - writer.WriteStartObject(); - - writer.WritePropertyName("_aws"); - writer.WriteStartObject(); - writer.WriteNumber("Timestamp", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()); - writer.WritePropertyName("CloudWatchMetrics"); - writer.WriteStartArray(); - writer.WriteStartObject(); - writer.WriteString("Namespace", Namespace); - writer.WritePropertyName("Dimensions"); - writer.WriteStartArray(); - writer.WriteStartArray(); - writer.WriteEndArray(); - writer.WriteEndArray(); - writer.WritePropertyName("Metrics"); - writer.WriteStartArray(); - foreach (var (name, _) in Counters) + Aws = new EmfEnvelope { - writer.WriteStartObject(); - writer.WriteString("Name", name); - writer.WriteString("Unit", "Count"); - writer.WriteEndObject(); - } - writer.WriteEndArray(); - writer.WriteEndObject(); - writer.WriteEndArray(); - writer.WriteEndObject(); + Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), + CloudWatchMetrics = + [ + new EmfMetricDirective + { + Namespace = Namespace, + Dimensions = [[]], + Metrics = MetricDefinitions + } + ] + }, + ObjectReconciles = metrics.ObjectReconciles, + ObjectReconcileRetries = metrics.ObjectReconcileRetries, + GroupReconciles = metrics.GroupReconciles, + RegistryWrites = metrics.RegistryWrites, + RegistryDeletes = metrics.RegistryDeletes, + RegistryUnchanged = metrics.RegistryUnchanged, + WriteConflicts = metrics.WriteConflicts, + ObjectsListed = metrics.ObjectsListed, + EntriesRecomputed = metrics.EntriesRecomputed, + ShallowRegistryWrites = metrics.ShallowRegistryWrites, + ShallowRegistryUnchanged = metrics.ShallowRegistryUnchanged, + FailedMessages = metrics.FailedMessages + }; + + Console.WriteLine(JsonSerializer.Serialize(payload, EmfJsonContext.Default.EmfPayload)); + } +} - foreach (var (name, value) in Counters) - writer.WriteNumber(name, value(metrics)); +/// One EMF log line: the _aws envelope plus the metric values as top-level members. +internal sealed record EmfPayload +{ + [JsonPropertyName("_aws")] + public required EmfEnvelope Aws { get; init; } - writer.WriteEndObject(); - } + public required int ObjectReconciles { get; init; } + public required int ObjectReconcileRetries { get; init; } + public required int GroupReconciles { get; init; } + public required int RegistryWrites { get; init; } + public required int RegistryDeletes { get; init; } + public required int RegistryUnchanged { get; init; } + public required int WriteConflicts { get; init; } + public required int ObjectsListed { get; init; } + public required int EntriesRecomputed { get; init; } + public required int ShallowRegistryWrites { get; init; } + public required int ShallowRegistryUnchanged { get; init; } + public required int FailedMessages { get; init; } +} - Console.WriteLine(Encoding.UTF8.GetString(buffer.ToArray())); - } +internal sealed record EmfEnvelope +{ + public required long Timestamp { get; init; } + public required IReadOnlyList CloudWatchMetrics { get; init; } } + +internal sealed record EmfMetricDirective +{ + public required string Namespace { get; init; } + public required IReadOnlyList> Dimensions { get; init; } + public required IReadOnlyList Metrics { get; init; } +} + +internal sealed record EmfMetricDefinition +{ + public required string Name { get; init; } + public string Unit { get; init; } = "Count"; +} + +[JsonSerializable(typeof(EmfPayload))] +internal sealed partial class EmfJsonContext : JsonSerializerContext; diff --git a/src/infra/docs-lambda-changelog-scrubber/Program.cs b/src/infra/docs-lambda-changelog-scrubber/Program.cs index c1f668129c..775f12a4af 100644 --- a/src/infra/docs-lambda-changelog-scrubber/Program.cs +++ b/src/infra/docs-lambda-changelog-scrubber/Program.cs @@ -55,7 +55,8 @@ async Task Handler(SQSEvent ev, ILambdaContext context) var metrics = new ReconcileMetrics(); var scrubber = new ChangelogContentScrubber(logFactory, allowRepos); var reconciler = new RegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); - var processor = new ScrubberProcessor(logFactory, s3Client, publicBucketName, scrubber, reconciler, metrics); + var shallowReconciler = new ShallowRegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); + var processor = new ScrubberProcessor(logFactory, s3Client, publicBucketName, scrubber, reconciler, shallowReconciler, metrics); var messages = ev.Records.Select(r => new ScrubberQueueMessage(r.MessageId, r.Body)).ToList(); var failedIds = await processor.ProcessAsync(messages, CancellationToken.None); diff --git a/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs b/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs index 4d6eb2c70f..02c7aec72f 100644 --- a/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs +++ b/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs @@ -21,6 +21,8 @@ public sealed class ReconcileMetrics private int _writeConflicts; private int _objectsListed; private int _entriesRecomputed; + private int _shallowRegistryWrites; + private int _shallowRegistryUnchanged; private int _failedMessages; /// Object-level reconciles run (one per distinct key in the batch). @@ -50,6 +52,12 @@ public sealed class ReconcileMetrics /// Entries whose metadata was recomputed from a public YAML read (vs. ETag-reused). public int EntriesRecomputed => _entriesRecomputed; + /// Shallow per-tree maps written or deleted (bundle/registry.json / changelog/registry.json). + public int ShallowRegistryWrites => _shallowRegistryWrites; + + /// Shallow-map reconciles that found the map already exact (steady state). + public int ShallowRegistryUnchanged => _shallowRegistryUnchanged; + /// SQS messages reported as batch-item failures. public int FailedMessages => _failedMessages; @@ -62,5 +70,7 @@ public sealed class ReconcileMetrics internal void IncrementWriteConflicts() => Interlocked.Increment(ref _writeConflicts); internal void IncrementObjectsListed() => Interlocked.Increment(ref _objectsListed); internal void IncrementEntriesRecomputed() => Interlocked.Increment(ref _entriesRecomputed); + internal void IncrementShallowRegistryWrites() => Interlocked.Increment(ref _shallowRegistryWrites); + internal void IncrementShallowRegistryUnchanged() => Interlocked.Increment(ref _shallowRegistryUnchanged); internal void AddFailedMessages(int count) => Interlocked.Add(ref _failedMessages, count); } diff --git a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs b/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs index a1148ae8f9..05a3d6a89f 100644 --- a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs +++ b/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs @@ -37,13 +37,20 @@ public enum GroupReconcileOutcome public sealed class ReconcileConflictException(string message) : Exception(message); /// -/// Rebuilds one group's public registry.json from the current public bucket state -/// (registry = f(state), never f(event) — see elastic/docs-eng-team#688). Lists the -/// group's prefix, reuses entries whose recorded ETag still matches, recomputes the rest from the -/// scrubbed public YAMLs, and writes the manifest back with optimistic concurrency. Any successful -/// reconcile therefore repairs all accumulated drift in the group, not just the change -/// that triggered it. +/// Rebuilds one bundle group's public registry.json from the current public bucket +/// state (registry = f(state), never f(event) — see elastic/docs-eng-team#688). +/// Lists the group's prefix, reuses entries whose recorded ETag still matches, recomputes the rest +/// from the scrubbed public YAMLs, and writes the manifest back with optimistic concurrency. Any +/// successful reconcile therefore repairs all accumulated drift in the group, not just +/// the change that triggered it. /// +/// +/// Scoped to the bundle/{product}/ tree only: the {changelog} directive and external +/// CDN consumers need to enumerate bundles, and dates (serverless) are not derivable client-side. +/// The changelog/… pool manifests are deliberately not reconciled — release-note +/// discovery starts from PR lists, so those manifests stay client-authored pass-through until +/// Phase 3 retires them entirely. +/// public sealed class RegistryReconciler( ILoggerFactory logFactory, IAmazonS3 s3Client, @@ -78,6 +85,9 @@ public sealed class RegistryReconciler( /// public async Task ReconcileGroupAsync(ChangelogScope scope, Cancel ctx) { + if (scope.Kind != ChangelogScopeKind.Bundle) + throw new ArgumentException($"Group reconcile applies to the bundle tree only; got '{scope}'.", nameof(scope)); + _metrics.IncrementGroupReconciles(); for (var attempt = 1; attempt <= MaxWriteAttempts; attempt++) @@ -230,19 +240,6 @@ private async Task FetchManifest(string key, Cancel ctx) IReadOnlyList reusable, Cancel ctx) { - // The entry index is listing-only by design (target is null; consumers re-read each entry), - // so the whole build is zero GETs. - if (scope.Kind == ChangelogScopeKind.Changelog) - { - var entries = Sort(listing.Select(obj => new RegistryBundle - { - File = obj.Key[scope.Prefix.Length..], - Target = null, - ETag = NormalizeETag(obj.ETag) - })); - return (entries, entries.Count); - } - var byFile = reusable.ToDictionary(b => b.File, b => b, StringComparer.Ordinal); var built = new RegistryBundle?[listing.Count]; var reused = 0; diff --git a/src/services/Elastic.Changelog/Reconciliation/ShallowRegistryReconciler.cs b/src/services/Elastic.Changelog/Reconciliation/ShallowRegistryReconciler.cs new file mode 100644 index 0000000000..52b310ce0a --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ShallowRegistryReconciler.cs @@ -0,0 +1,341 @@ +// 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 System.Net; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using System.Text.Json.Serialization; +using Amazon.S3; +using Amazon.S3.Model; +using Elastic.Documentation.Configuration.ReleaseNotes; +using Microsoft.Extensions.Logging; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// Maintains the shallow per-tree registries on the public bucket — +/// bundle/registry.json and changelog/registry.json — mapping each folder (a +/// product, or an {org}/{repo}/{branch} pool) to an opaque change token. CDN consumers +/// that cache a folder's content can compare one small object to decide whether anything under +/// that folder changed, before diving into the folder itself. +/// +/// +/// +/// The token is a digest over the folder's full listing (sorted file/ETag pairs), not the ETag of +/// any single object: a "last-touched object's ETag" goes stale when an older object is +/// deleted, since the newest object — and therefore the value — would not change. Consumers must +/// treat the value as opaque. +/// +/// +/// Like the group reconciler, this is f(state): touched folders are re-listed and the map +/// is patched with optimistic concurrency. An absent or unparseable map is rebuilt from a full +/// tree listing, which is also how the map is seeded on first deploy. +/// +/// +public sealed class ShallowRegistryReconciler( + ILoggerFactory logFactory, + IAmazonS3 s3Client, + string publicBucketName, + TimeSpan? retryBaseDelay = null, + ReconcileMetrics? metrics = null +) +{ + // Bounds the optimistic-concurrency retry loop; each attempt re-lists and re-reads before retrying. + private const int MaxWriteAttempts = 5; + + private readonly ILogger _logger = logFactory.CreateLogger(); + private readonly TimeSpan _retryBaseDelay = retryBaseDelay ?? TimeSpan.FromMilliseconds(200); + private readonly ReconcileMetrics _metrics = metrics ?? new ReconcileMetrics(); + + /// + /// Converges the tree's shallow map for folders (all of + /// ). Throws when concurrent + /// conditional writers win every bounded retry — the caller fails the SQS message so + /// redelivery retries later. + /// + public async Task ReconcileAsync(ChangelogScopeKind kind, IReadOnlyCollection touched, Cancel ctx) + { + if (touched.Count == 0) + return; + if (touched.Any(scope => scope.Kind != kind)) + throw new ArgumentException($"Every touched scope must be of kind {kind}.", nameof(touched)); + + var mapKey = TreeRegistryKey(kind); + + for (var attempt = 1; attempt <= MaxWriteAttempts; attempt++) + { + ctx.ThrowIfCancellationRequested(); + + var existing = await FetchMap(mapKey, ctx); + + SortedDictionary map; + if (existing.Map is { } parsed) + { + map = parsed; + foreach (var scope in touched) + { + var token = await ComputeFolderToken(scope, ctx); + if (token is null) + _ = map.Remove(scope.Group); + else + map[scope.Group] = token; + } + } + else + { + // Absent or unparseable: rebuild the whole tree's map from one full listing. This + // is the first-deploy seed path too, so a single event heals every folder at once. + map = await RebuildTreeMap(kind, ctx); + } + + if (existing.Map is not null && MapsEqual(existing.Map, existing.Original!, map)) + { + _metrics.IncrementShallowRegistryUnchanged(); + _logger.LogDebug("Shallow map {Key} already matches state; skipping write", mapKey); + return; + } + + var converged = map.Count == 0 + ? await TryDeleteMap(mapKey, existing.ETag, attempt, ctx) + : await TryPutMap(mapKey, map, existing.ETag, attempt, ctx); + if (converged) + return; + await BackOff(attempt, ctx); + } + + throw new ReconcileConflictException( + $"Shallow map {mapKey} kept changing concurrently after {MaxWriteAttempts} attempts; failing the message for redelivery."); + } + + /// The S3 key of a tree's shallow map: bundle/registry.json or changelog/registry.json. + public static string TreeRegistryKey(ChangelogScopeKind kind) => kind == ChangelogScopeKind.Bundle + ? $"{ChangelogKeys.BundlePrefix}{ChangelogKeys.RegistryFileName}" + : $"{ChangelogKeys.ChangelogPrefix}{ChangelogKeys.RegistryFileName}"; + + /// + /// The folder's change token from its current public listing, or null when the folder holds no + /// content. Group manifests are excluded: they are derived from the same content this token + /// already covers, and the reconciler rewriting one must not invalidate consumer caches. + /// + private async Task ComputeFolderToken(ChangelogScope scope, Cancel ctx) + { + var files = new SortedDictionary(StringComparer.Ordinal); + var request = new ListObjectsV2Request + { + BucketName = publicBucketName, + Prefix = scope.Prefix, + Delimiter = "/" + }; + + ListObjectsV2Response response; + do + { + response = await s3Client.ListObjectsV2Async(request, ctx); + foreach (var obj in response.S3Objects ?? []) + { + var file = obj.Key[scope.Prefix.Length..]; + if (IsYamlFileName(file)) + files[file] = NormalizeETag(obj.ETag); + } + request.ContinuationToken = response.NextContinuationToken; + } while (response.IsTruncated == true); + + return files.Count == 0 ? null : TokenOf(files); + } + + /// + /// Rebuilds every folder's token of from one full (undelimited) tree + /// listing. Keys that do not parse into a valid scope, nested keys, and manifests are skipped — + /// the same content rules the per-folder listing applies. + /// + private async Task> RebuildTreeMap(ChangelogScopeKind kind, Cancel ctx) + { + var folders = new Dictionary>(StringComparer.Ordinal); + var request = new ListObjectsV2Request + { + BucketName = publicBucketName, + Prefix = kind == ChangelogScopeKind.Bundle ? ChangelogKeys.BundlePrefix : ChangelogKeys.ChangelogPrefix + }; + + ListObjectsV2Response response; + do + { + response = await s3Client.ListObjectsV2Async(request, ctx); + foreach (var obj in response.S3Objects ?? []) + { + if (!ChangelogScope.TryFromKey(obj.Key, out var scope) || scope.Kind != kind) + continue; + + var file = obj.Key[scope.Prefix.Length..]; + if (!IsYamlFileName(file) || file.Contains('/', StringComparison.Ordinal)) + continue; + + if (!folders.TryGetValue(scope.Group, out var files)) + { + files = [with(StringComparer.Ordinal)]; + folders[scope.Group] = files; + } + files[file] = NormalizeETag(obj.ETag); + } + request.ContinuationToken = response.NextContinuationToken; + } while (response.IsTruncated == true); + + var map = new SortedDictionary(StringComparer.Ordinal); + foreach (var (group, files) in folders) + map[group] = TokenOf(files); + return map; + } + + private static bool IsYamlFileName(string file) => + file.Length > 0 + && (file.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase) || file.EndsWith(".yml", StringComparison.OrdinalIgnoreCase)); + + private static string TokenOf(SortedDictionary files) + { + var builder = new StringBuilder(); + foreach (var (file, etag) in files) + _ = builder.Append(file).Append('\n').Append(etag).Append('\n'); + return Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(builder.ToString())))[..32]; + } + + private sealed record MapState(SortedDictionary? Map, string? Original, string? ETag); + + /// Reads the map, distinguishing absent (no ETag) from unparseable (live ETag, null map). + private async Task FetchMap(string key, Cancel ctx) + { + string? etag = null; + try + { + using var response = await s3Client.GetObjectAsync(new GetObjectRequest + { + BucketName = publicBucketName, + Key = key + }, ctx); + + etag = response.ETag; + await using var stream = response.ResponseStream; + using var reader = new StreamReader(stream); + var original = await reader.ReadToEndAsync(ctx); + var map = JsonSerializer.Deserialize(original, ShallowRegistryJsonContext.Default.SortedDictionaryStringString); + return new MapState(map, original, etag); + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return new MapState(null, null, null); + } + catch (JsonException ex) + { + // An unparseable map is rebuilt from the tree listing; its live ETag lets the + // conditional write replace it safely. + _logger.LogWarning(ex, "Shallow map {Key} could not be parsed; rebuilding from the tree listing", key); + return new MapState(null, null, etag); + } + } + + /// + /// Serialized-form comparison: a map that parses to the same pairs but was not written by this + /// serializer (different ordering/whitespace) is rewritten once so the stored bytes converge. + /// + private static bool MapsEqual(SortedDictionary before, string original, SortedDictionary after) + { + if (before.Count != after.Count) + return false; + + foreach (var (group, token) in after) + { + if (!before.TryGetValue(group, out var existing) || !string.Equals(existing, token, StringComparison.Ordinal)) + return false; + } + + return string.Equals(original, Serialize(after), StringComparison.Ordinal); + } + + private static string Serialize(SortedDictionary map) => + JsonSerializer.Serialize(map, ShallowRegistryJsonContext.Default.SortedDictionaryStringString); + + private async Task TryPutMap(string key, SortedDictionary map, string? etag, int attempt, Cancel ctx) + { + var request = new PutObjectRequest + { + BucketName = publicBucketName, + Key = key, + ContentBody = Serialize(map), + ContentType = "application/json" + }; + + // Optimistic concurrency: update only if unchanged, create only if still absent. + if (etag is null) + request.IfNoneMatch = "*"; + else + request.IfMatch = etag; + + try + { + _ = await s3Client.PutObjectAsync(request, ctx); + _metrics.IncrementShallowRegistryWrites(); + _logger.LogInformation("Wrote shallow map {Key} with {Count} folder(s)", key, map.Count); + return true; + } + catch (AmazonS3Exception ex) when (IsConditionalWriteConflict(ex)) + { + _metrics.IncrementWriteConflicts(); + _logger.LogInformation( + "Shallow map {Key} changed concurrently (attempt {Attempt}/{Max}); re-listing and retrying", + key, attempt, MaxWriteAttempts); + return false; + } + } + + private async Task TryDeleteMap(string key, string? etag, int attempt, Cancel ctx) + { + if (etag is null) + return true; + + try + { + _ = await s3Client.DeleteObjectAsync(new DeleteObjectRequest + { + BucketName = publicBucketName, + Key = key, + IfMatch = etag + }, ctx); + _metrics.IncrementShallowRegistryWrites(); + _logger.LogInformation("Deleted shallow map {Key}: the tree is empty", key); + return true; + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + // Another reconciler removed it first; converged all the same. + return true; + } + catch (AmazonS3Exception ex) when (IsConditionalWriteConflict(ex)) + { + _metrics.IncrementWriteConflicts(); + _logger.LogInformation( + "Shallow map {Key} changed concurrently during delete (attempt {Attempt}/{Max}); re-listing and retrying", + key, attempt, MaxWriteAttempts); + return false; + } + } + + // 412 = a plain conditional-request loss; 409 = ConditionalRequestConflict, S3's signal for + // concurrent conditional writers on the same key. Both mean: re-read state and retry. + private static bool IsConditionalWriteConflict(AmazonS3Exception ex) => + ex.StatusCode is HttpStatusCode.PreconditionFailed or HttpStatusCode.Conflict; + + private async Task BackOff(int attempt, Cancel ctx) + { + if (_retryBaseDelay <= TimeSpan.Zero) + return; + var jitter = TimeSpan.FromMilliseconds(Random.Shared.NextDouble() * _retryBaseDelay.TotalMilliseconds); + await Task.Delay((_retryBaseDelay * attempt) + jitter, ctx); + } + + private static string NormalizeETag(string? etag) => etag?.Trim('"') ?? string.Empty; +} + +[JsonSourceGenerationOptions(WriteIndented = true)] +[JsonSerializable(typeof(SortedDictionary))] +public sealed partial class ShallowRegistryJsonContext : JsonSerializerContext; diff --git a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs index a70af45507..0de6ee1072 100644 --- a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs +++ b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs @@ -19,8 +19,9 @@ public sealed record ScrubberQueueMessage(string MessageId, string Body); /// The scrubber Lambda's event processor (elastic/docs-eng-team#688), extracted from /// Program.cs so it is testable. Events are triggers, state decides: the handler never /// acts on an event's type — an event means only "this key may have changed, look at -/// it". Every distinct key gets one object-level reconcile against the private bucket, then every -/// distinct group gets one registry reconcile against the public listing, so out-of-order and +/// it". Every distinct key gets one object-level reconcile against the private bucket; every +/// distinct bundle/{product}/ group then gets one registry reconcile against the public +/// listing, and every touched tree gets one shallow-map reconcile. Out-of-order and /// at-least-once S3 notifications are harmless and each batch heals accumulated drift. /// public sealed class ScrubberProcessor( @@ -29,6 +30,7 @@ public sealed class ScrubberProcessor( string publicBucketName, IChangelogContentScrubber scrubber, RegistryReconciler reconciler, + ShallowRegistryReconciler shallowReconciler, ReconcileMetrics? metrics = null ) { @@ -40,9 +42,13 @@ public sealed class ScrubberProcessor( private readonly ILogger _logger = logFactory.CreateLogger(); private readonly ReconcileMetrics _metrics = metrics ?? new ReconcileMetrics(); - private sealed class ObjectWork(string sourceBucket) + private sealed class ObjectWork(string sourceBucket, bool passThrough) { public string SourceBucket { get; set; } = sourceBucket; + + /// True for a pool manifest copied verbatim; false for YAML content that is scrubbed. + public bool PassThrough { get; } = passThrough; + public HashSet MessageIds { get; } = [with(StringComparer.Ordinal)]; } @@ -52,6 +58,13 @@ private sealed class GroupWork(ChangelogScope scope) public HashSet MessageIds { get; } = [with(StringComparer.Ordinal)]; } + private sealed class ShallowWork(ChangelogScopeKind kind) + { + public ChangelogScopeKind Kind { get; } = kind; + public Dictionary Scopes { get; } = [with(StringComparer.Ordinal)]; + public HashSet MessageIds { get; } = [with(StringComparer.Ordinal)]; + } + /// /// Processes one SQS batch and returns the message ids that must be redelivered. Work is /// coalesced per distinct key and per distinct group; a failed object reconcile fails every @@ -62,6 +75,7 @@ public async Task> ProcessAsync(IReadOnlyList(StringComparer.Ordinal); var groupWork = new Dictionary(StringComparer.Ordinal); + var shallowWork = new Dictionary(); var failedIds = new HashSet(StringComparer.Ordinal); foreach (var message in messages) @@ -73,7 +87,7 @@ public async Task> ProcessAsync(IReadOnlyList> ProcessAsync(IReadOnlyList> ProcessAsync(IReadOnlyList objectWork, - Dictionary groupWork) + Dictionary groupWork, + Dictionary shallowWork) { var hasScope = ChangelogScope.TryFromKey(key, out var scope); - // The registry pass-through is retired: registry keys are never copied or deleted. Old CLI - // versions still write private manifests (and Phase 3's cleanup will delete them) — those - // events only schedule a reconcile of the group, which derives the public manifest from - // public state. if (ChangelogKeys.IsRegistry(key)) { - if (hasScope) - AddGroup(groupWork, scope!, messageId); + if (!hasScope) + return; + + // The two trees part ways here. Bundle manifests are reconciler-owned: the event only + // schedules a group reconcile, so client-authored JSON never reaches the public bucket + // for the tree consumers enumerate. Pool manifests stay client-authored pass-through — + // `changelog bundle` still enumerates a pool through its manifest today, and 404-probing + // only works once entries are guaranteed one-per-PR — until Phase 3 retires them. + if (scope!.Kind == ChangelogScopeKind.Bundle) + AddGroup(groupWork, scope, messageId); + else + AddObject(objectWork, key, sourceBucket, messageId, passThrough: true); return; } @@ -148,16 +183,30 @@ private void Classify( return; } + AddObject(objectWork, key, sourceBucket, messageId, passThrough: false); + + if (!hasScope) + return; + + if (scope!.Kind == ChangelogScopeKind.Bundle) + AddGroup(groupWork, scope, messageId); + AddShallow(shallowWork, scope, messageId); + } + + private static void AddObject( + Dictionary objectWork, + string key, + string sourceBucket, + string messageId, + bool passThrough) + { if (!objectWork.TryGetValue(key, out var work)) { - work = new ObjectWork(sourceBucket); + work = new ObjectWork(sourceBucket, passThrough); objectWork[key] = work; } work.SourceBucket = sourceBucket; _ = work.MessageIds.Add(messageId); - - if (hasScope) - AddGroup(groupWork, scope!, messageId); } private static void AddGroup(Dictionary groupWork, ChangelogScope scope, string messageId) @@ -170,12 +219,24 @@ private static void AddGroup(Dictionary groupWork, ChangelogS _ = work.MessageIds.Add(messageId); } + private static void AddShallow(Dictionary shallowWork, ChangelogScope scope, string messageId) + { + if (!shallowWork.TryGetValue(scope.Kind, out var work)) + { + work = new ShallowWork(scope.Kind); + shallowWork[scope.Kind] = work; + } + work.Scopes[scope.Group] = scope; + _ = work.MessageIds.Add(messageId); + } + /// /// Order-independent object reconcile: the event type is ignored; the private bucket's current - /// state decides between scrub-and-copy and delete. A stale ObjectRemoved arriving - /// after a recreate re-copies the live object instead of deleting it. + /// state decides between copy and delete. A stale ObjectRemoved arriving after a + /// recreate re-copies the live object instead of deleting it. YAML content is scrubbed on the + /// way through; a pass-through pool manifest is copied verbatim. /// - private async Task ReconcileObjectAsync(string sourceBucket, string key, Cancel ctx) + private async Task ReconcileObjectAsync(string sourceBucket, string key, bool passThrough, Cancel ctx) { _metrics.IncrementObjectReconciles(); @@ -186,9 +247,17 @@ private async Task ReconcileObjectAsync(string sourceBucket, string key, Cancel var snapshot = await TryGetPrivateObject(sourceBucket, key, ctx); if (snapshot is { } source) { - var scrubbed = await scrubber.ScrubAsync(key, source.Content, ctx); - await PutPublicObject(key, scrubbed, ctx); - _logger.LogInformation("Scrubbed and wrote {Key} to public bucket", key); + if (passThrough) + { + await PutPublicObject(key, source.Content, "application/json", ctx); + _logger.LogInformation("Copied {Key} to public bucket (pass-through)", key); + } + else + { + var scrubbed = await scrubber.ScrubAsync(key, source.Content, ctx); + await PutPublicObject(key, scrubbed, "application/yaml", ctx); + _logger.LogInformation("Scrubbed and wrote {Key} to public bucket", key); + } } else { @@ -256,13 +325,13 @@ private async Task ReconcileObjectAsync(string sourceBucket, string key, Cancel } } - private async Task PutPublicObject(string key, string content, Cancel ctx) => + private async Task PutPublicObject(string key, string content, string contentType, Cancel ctx) => _ = await s3Client.PutObjectAsync(new PutObjectRequest { BucketName = publicBucketName, Key = key, ContentBody = content, - ContentType = "application/yaml" + ContentType = contentType }, ctx); private async Task DeletePublicObject(string key, Cancel ctx) diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs index 607212e2d2..90c90c39c2 100644 --- a/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs +++ b/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs @@ -268,37 +268,17 @@ public async Task ReconcileGroup_ListingPaginates() } [Fact] - public async Task ReconcileGroup_BranchWithSlash_DoesNotSweepNestedPools() - { - // The "main" pool must not swallow the "main/feature" pool: branches are stored verbatim, - // so the / delimiter is what keeps them apart. - var scope = ChangelogScopeFor("elastic", "repo", "main"); - _ = _s3.Seed(PublicBucket, "changelog/elastic/repo/main/entry-a.yaml", "a: 1"); - _ = _s3.Seed(PublicBucket, "changelog/elastic/repo/main/feature/entry-b.yaml", "b: 1"); - - var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); - - outcome.Should().Be(GroupReconcileOutcome.Written); - var content = _s3.ContentOf(PublicBucket, scope.RegistryKey); - var manifest = JsonSerializer.Deserialize(content, RegistryJsonContext.Default.Registry)!; - manifest.Bundles.Select(b => b.File).Should().Equal("entry-a.yaml"); - } - - [Fact] - public async Task ReconcileGroup_ChangelogScope_IsListingOnly_ZeroObjectReads() + public async Task ReconcileGroup_ChangelogScope_IsRejected() { + // Pool manifests are not reconciled: they stay client-authored pass-through until Phase 3 + // retires them, so a changelog scope reaching the group reconciler is a programming error. var scope = ChangelogScopeFor("elastic", "repo", "main"); _ = _s3.Seed(PublicBucket, scope.Prefix + "entry-a.yaml", "a: 1"); - _ = _s3.Seed(PublicBucket, scope.Prefix + "entry-b.yaml", "b: 1"); - var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); + var act = async () => await _reconciler.ReconcileGroupAsync(scope, Ctx); - outcome.Should().Be(GroupReconcileOutcome.Written); - var content = _s3.ContentOf(PublicBucket, scope.RegistryKey); - var manifest = JsonSerializer.Deserialize(content, RegistryJsonContext.Default.Registry)!; - manifest.Bundles.Should().HaveCount(2); - manifest.Bundles.Should().OnlyContain(b => b.Target == null, "the entry index is listing-only by design"); - _s3.GetsFor(PublicBucket).Should().Equal(scope.RegistryKey); + _ = await act.Should().ThrowAsync(); + _s3.Puts.Should().BeEmpty(); } [Fact] diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/ShallowRegistryReconcilerTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/ShallowRegistryReconcilerTests.cs new file mode 100644 index 0000000000..b9e071ec37 --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Reconciliation/ShallowRegistryReconcilerTests.cs @@ -0,0 +1,216 @@ +// 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 System.Text.Json; +using AwesomeAssertions; +using Elastic.Changelog.Reconciliation; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Elastic.Changelog.Tests.Reconciliation; + +public class ShallowRegistryReconcilerTests +{ + private const string PublicBucket = "public-bucket"; + private const string BundleMapKey = "bundle/registry.json"; + private const string ChangelogMapKey = "changelog/registry.json"; + + private readonly FakeS3 _s3 = new(PublicBucket); + private readonly ReconcileMetrics _metrics = new(); + private readonly ShallowRegistryReconciler _reconciler; + + public ShallowRegistryReconcilerTests() => + _reconciler = new ShallowRegistryReconciler( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); + + private Cancel Ctx => TestContext.Current.CancellationToken; + + private static ChangelogScope BundleScope(string product) + { + _ = ChangelogScope.TryCreateBundle(product, out var scope); + return scope!; + } + + private static ChangelogScope PoolScope(string org, string repo, string branch) + { + _ = ChangelogScope.TryCreateChangelog(org, repo, branch, out var scope); + return scope!; + } + + private SortedDictionary Map(string key) => + JsonSerializer.Deserialize(_s3.ContentOf(PublicBucket, key), ShallowRegistryJsonContext.Default.SortedDictionaryStringString)!; + + private void SeedMap(string key, params (string Group, string Token)[] entries) + { + var map = new SortedDictionary(StringComparer.Ordinal); + foreach (var (group, token) in entries) + map[group] = token; + _ = _s3.Seed(PublicBucket, key, JsonSerializer.Serialize(map, ShallowRegistryJsonContext.Default.SortedDictionaryStringString)); + } + + [Fact] + public async Task Reconcile_AbsentMap_SeedsEveryFolderFromAFullTreeListing() + { + // First deploy: no map exists yet, so a single touched folder heals the whole tree. + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PublicBucket, "bundle/kibana/kb-9.1.0.yaml", "two"); + _ = _s3.Seed(PublicBucket, "bundle/kibana/registry.json", "{}"); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch")], Ctx); + + Map(BundleMapKey).Keys.Should().BeEquivalentTo("elasticsearch", "kibana"); + } + + [Fact] + public async Task Reconcile_ExistingMap_PatchesOnlyTheTouchedFolders() + { + // Untouched folders keep their recorded value even when stale — their own events (or the + // next absent-map rebuild) heal them. Patching avoids listing the whole tree per event. + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PublicBucket, "bundle/kibana/kb-9.1.0.yaml", "two"); + SeedMap(BundleMapKey, ("kibana", "stale-token")); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch")], Ctx); + + var map = Map(BundleMapKey); + map.Should().ContainKey("elasticsearch"); + map["kibana"].Should().Be("stale-token"); + } + + [Fact] + public async Task Reconcile_TokenChangesWhenAnyFileChangesOrIsDeleted() + { + // The token digests the whole listing rather than echoing one object's ETag: deleting an + // older file must change the value, or opt-out caches would never see the delete. + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "old"); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.2.0.yaml", "new"); + var scope = BundleScope("elasticsearch"); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [scope], Ctx); + var before = Map(BundleMapKey)["elasticsearch"]; + + _s3.Remove(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml"); + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [scope], Ctx); + + Map(BundleMapKey)["elasticsearch"].Should().NotBe(before); + } + + [Fact] + public async Task Reconcile_UnchangedFolder_SkipsTheWrite() + { + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + var scope = BundleScope("elasticsearch"); + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [scope], Ctx); + var puts = _s3.Puts.Count; + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [scope], Ctx); + + _s3.Puts.Count.Should().Be(puts, "an exact map must not be rewritten"); + _metrics.ShallowRegistryUnchanged.Should().Be(1); + } + + [Fact] + public async Task Reconcile_EmptiedFolder_IsRemovedFromTheMap() + { + _ = _s3.Seed(PublicBucket, "bundle/kibana/kb-9.1.0.yaml", "keep"); + SeedMap(BundleMapKey, ("elasticsearch", "token"), ("kibana", "token")); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch"), BundleScope("kibana")], Ctx); + + Map(BundleMapKey).Keys.Should().BeEquivalentTo("kibana"); + } + + [Fact] + public async Task Reconcile_EmptiedTree_DeletesTheMapConditionally() + { + SeedMap(BundleMapKey, ("elasticsearch", "token")); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch")], Ctx); + + _s3.Exists(PublicBucket, BundleMapKey).Should().BeFalse("an empty tree's map is deleted: absent ≠ empty"); + _s3.Deletes.Should().ContainSingle().Which.IfMatch.Should().NotBeNullOrEmpty(); + } + + [Fact] + public async Task Reconcile_PoolBranchWithSlash_DoesNotSweepNestedPools() + { + // Branches are stored verbatim, so the / delimiter is what keeps the "main" pool from + // swallowing the "main/feature" pool's files into its token. + _ = _s3.Seed(PublicBucket, "changelog/elastic/repo/main/entry-a.yaml", "a"); + _ = _s3.Seed(PublicBucket, "changelog/elastic/repo/main/feature/entry-b.yaml", "b"); + SeedMap(ChangelogMapKey, ("elastic/repo/main", "seed"), ("elastic/repo/main/feature", "seed")); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Changelog, [PoolScope("elastic", "repo", "main")], Ctx); + var afterParent = Map(ChangelogMapKey)["elastic/repo/main"]; + await _reconciler.ReconcileAsync(ChangelogScopeKind.Changelog, [PoolScope("elastic", "repo", "main/feature")], Ctx); + + var map = Map(ChangelogMapKey); + map["elastic/repo/main"].Should().Be(afterParent); + map["elastic/repo/main/feature"].Should().NotBe("seed").And.NotBe(afterParent); + } + + [Fact] + public async Task Reconcile_UnparseableMap_IsRebuiltFromTheTreeListing() + { + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PublicBucket, "bundle/kibana/kb-9.1.0.yaml", "two"); + var corruptETag = _s3.Seed(PublicBucket, BundleMapKey, "{ not json "); + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch")], Ctx); + + Map(BundleMapKey).Keys.Should().BeEquivalentTo("elasticsearch", "kibana"); + _s3.Puts.Should().ContainSingle().Which.IfMatch.Trim('"').Should().Be(corruptETag, + "the conditional write must replace exactly the corrupt map that was read"); + } + + [Fact] + public async Task Reconcile_PutLosingTheRace_RereadsAndRetries() + { + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + var raced = false; + _s3.BeforePut = _ => + { + if (raced) + return; + raced = true; + SeedMap(BundleMapKey, ("kibana", "concurrent")); + }; + + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch")], Ctx); + + _metrics.WriteConflicts.Should().Be(1); + var map = Map(BundleMapKey); + map.Should().ContainKey("elasticsearch"); + map.Should().ContainKey("kibana", "the concurrent writer's entry was re-read and preserved"); + } + + [Fact] + public async Task Reconcile_ExhaustedConditionalRetries_Throws() + { + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + var counter = 0; + _s3.BeforePut = _ => SeedMap(BundleMapKey, ("racer", $"token-{counter++}")); + + var act = async () => await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [BundleScope("elasticsearch")], Ctx); + + _ = await act.Should().ThrowAsync(); + } + + [Fact] + public async Task Reconcile_NoTouchedFolders_IsANoOp() + { + await _reconciler.ReconcileAsync(ChangelogScopeKind.Bundle, [], Ctx); + + _s3.ListCalls.Should().Be(0); + _s3.Puts.Should().BeEmpty(); + } + + [Fact] + public async Task Reconcile_MismatchedScopeKind_IsRejected() + { + var act = async () => await _reconciler.ReconcileAsync( + ChangelogScopeKind.Bundle, [PoolScope("elastic", "repo", "main")], Ctx); + + _ = await act.Should().ThrowAsync(); + } +} diff --git a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs index 5fb95d40b6..e0a10cb2a4 100644 --- a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs +++ b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs @@ -32,8 +32,10 @@ public ScrubberProcessorTests() var reconciler = new RegistryReconciler( NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); + var shallowReconciler = new ShallowRegistryReconciler( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); _processor = new ScrubberProcessor( - NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, _metrics); + NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, shallowReconciler, _metrics); } private Cancel Ctx => TestContext.Current.CancellationToken; @@ -53,6 +55,9 @@ private static ScrubberQueueMessage Message(string eventName, string key, string private Registry PublicManifest(string registryKey) => JsonSerializer.Deserialize(_s3.ContentOf(PublicBucket, registryKey), RegistryJsonContext.Default.Registry)!; + private SortedDictionary ShallowMap(string mapKey) => + JsonSerializer.Deserialize(_s3.ContentOf(PublicBucket, mapKey), ShallowRegistryJsonContext.Default.SortedDictionaryStringString)!; + [Fact] public async Task Process_CreatedEvent_ScrubsCopiesAndWritesTheGroupManifest() { @@ -97,11 +102,12 @@ public async Task Process_StaleCreatedEventAfterDelete_RemovesThePublicCopyAndMa } [Fact] - public async Task Process_RegistryKeyEvents_NeverCopyOrDelete_OnlyTriggerAGroupReconcile() + public async Task Process_BundleRegistryKeyEvents_NeverCopyOrDelete_OnlyTriggerAGroupReconcile() { - // Old CLI versions still write private manifests; Phase 3's cleanup will delete them. - // Neither event may touch the public registry object directly — but each schedules a - // reconcile that derives the public manifest from public state. + // The bundle manifest is reconciler-owned. Old CLI versions still write private bundle + // manifests (and Phase 3's cleanup will delete them) — those events may never touch the + // public registry object directly, only schedule a reconcile that derives the public + // manifest from public state. _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/registry.json", /*lang=json,strict*/ """{"private":"manifest"}"""); _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", BundleYaml()); @@ -112,7 +118,85 @@ public async Task Process_RegistryKeyEvents_NeverCopyOrDelete_OnlyTriggerAGroupR var manifest = PublicManifest("bundle/elasticsearch/registry.json"); manifest.Producer.Should().Be(RegistryReconciler.Producer); manifest.Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); - _s3.GetsFor(PrivateBucket).Should().BeEmpty("the private registry content must never be read for pass-through"); + _s3.GetsFor(PrivateBucket).Should().BeEmpty("the private bundle registry content must never be read for pass-through"); + } + + [Fact] + public async Task Process_PoolRegistryKeyEvents_ArePassedThroughVerbatim() + { + // Pool manifests stay client-authored until Phase 3: `changelog bundle` still enumerates a + // pool through its manifest, so the private copy is mirrored verbatim — never scrubbed, + // never reconciled. + const string poolRegistry = "changelog/elastic/kibana/main/registry.json"; + const string content = /*lang=json,strict*/ """{"schema_version":1,"bundles":[{"file":"100.yaml"}]}"""; + _ = _s3.Seed(PrivateBucket, poolRegistry, content); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", poolRegistry)], Ctx); + + failed.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, poolRegistry).Should().Be(content, "pass-through must not transform the manifest"); + _metrics.GroupReconciles.Should().Be(0, "pool manifests are not reconciled"); + _s3.Puts.Single(p => p.Key == poolRegistry).ContentType.Should().Be("application/json"); + } + + [Fact] + public async Task Process_PoolRegistryKeyEvents_WithPrivateGone_DeleteThePublicCopy() + { + // State decides for pass-through keys too: Phase 3's private-manifest cleanup deletes will + // propagate and remove the public pool manifests with them. + const string poolRegistry = "changelog/elastic/kibana/main/registry.json"; + _ = _s3.Seed(PublicBucket, poolRegistry, "{}"); + + var failed = await _processor.ProcessAsync([Message("ObjectRemoved:Delete", poolRegistry)], Ctx); + + failed.Should().BeEmpty(); + _s3.Exists(PublicBucket, poolRegistry).Should().BeFalse(); + } + + [Fact] + public async Task Process_PoolYamlEvents_ScrubAndUpdateTheShallowMap_ButWriteNoPoolManifest() + { + _ = _s3.Seed(PrivateBucket, "changelog/elastic/kibana/main/100.yaml", "entry"); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "changelog/elastic/kibana/main/100.yaml")], Ctx); + + failed.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, "changelog/elastic/kibana/main/100.yaml").Should().Be("scrubbed: entry"); + _s3.Exists(PublicBucket, "changelog/elastic/kibana/main/registry.json") + .Should().BeFalse("the reconciler no longer produces pool manifests"); + _metrics.GroupReconciles.Should().Be(0); + + var map = ShallowMap("changelog/registry.json"); + map.Should().ContainKey("elastic/kibana/main"); + } + + [Fact] + public async Task Process_BundleYamlEvents_UpdateTheShallowMapForTheProduct() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "content"); + + var failed = await _processor.ProcessAsync([Message("ObjectCreated:Put", "bundle/elasticsearch/es-9.1.0.yaml")], Ctx); + + failed.Should().BeEmpty(); + var map = ShallowMap("bundle/registry.json"); + map.Should().ContainKey("elasticsearch"); + } + + [Fact] + public async Task Process_MultiplePoolsInOneBatch_CoalesceIntoASingleShallowMapWrite() + { + _ = _s3.Seed(PrivateBucket, "changelog/elastic/kibana/main/100.yaml", "one"); + _ = _s3.Seed(PrivateBucket, "changelog/elastic/elasticsearch/main/200.yaml", "two"); + + var failed = await _processor.ProcessAsync( + [ + Message("ObjectCreated:Put", "changelog/elastic/kibana/main/100.yaml"), + Message("ObjectCreated:Put", "changelog/elastic/elasticsearch/main/200.yaml") + ], Ctx); + + failed.Should().BeEmpty(); + _s3.Puts.Where(p => p.Key == "changelog/registry.json").Should().ContainSingle("one tree gets one map write per batch"); + ShallowMap("changelog/registry.json").Keys.Should().BeEquivalentTo("elastic/kibana/main", "elastic/elasticsearch/main"); } [Fact] @@ -254,8 +338,8 @@ public async Task Process_KeyOutsideAnyGroupLayout_IsCopiedButTriggersNoGroupRec [Fact] public async Task Process_BatchMixingObjectAndRegistryEvents_MarksGroupContributionsAcrossBoth() { - // A YAML event and a (retired pass-through) registry event for the same group coalesce - // into one group reconcile fed by both messages. + // A YAML event and a bundle-registry event for the same group coalesce into one group + // reconcile fed by both messages. _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/registry.json", "{}"); From b7fe622ac60120fd9e4c1564cf98128acc7f440b Mon Sep 17 00:00:00 2001 From: Felipe Cotti Date: Mon, 10 Aug 2026 14:08:57 -0300 Subject: [PATCH 3/3] Clarify per-tree registry ownership in code and docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rename RegistryReconciler to BundleRegistryReconciler: since the rework it is scoped to the bundle/{product}/ tree only, and the old name read as if it still covered the changelog pool manifests the review asked to drop. Fix Registry.cs doc comments that predate the rework: public bundle indexes are reconciler-owned (not mirrored verbatim), and their recorded ETag is now the public object's — valid for CDN cache validation — while client-authored manifests keep recording the private pre-scrub ETag. --- .../docs-lambda-changelog-scrubber/Program.cs | 2 +- ...onciler.cs => BundleRegistryReconciler.cs} | 4 +-- .../Scrubbing/ScrubberProcessor.cs | 2 +- .../Elastic.Changelog/Uploading/Registry.cs | 26 ++++++++++++------- ...ts.cs => BundleRegistryReconcilerTests.cs} | 16 ++++++------ .../Scrubbing/ScrubberProcessorTests.cs | 4 +-- 6 files changed, 30 insertions(+), 24 deletions(-) rename src/services/Elastic.Changelog/Reconciliation/{RegistryReconciler.cs => BundleRegistryReconciler.cs} (99%) rename tests/Elastic.Changelog.Tests/Reconciliation/{RegistryReconcilerTests.cs => BundleRegistryReconcilerTests.cs} (96%) diff --git a/src/infra/docs-lambda-changelog-scrubber/Program.cs b/src/infra/docs-lambda-changelog-scrubber/Program.cs index 775f12a4af..6edd17f89a 100644 --- a/src/infra/docs-lambda-changelog-scrubber/Program.cs +++ b/src/infra/docs-lambda-changelog-scrubber/Program.cs @@ -54,7 +54,7 @@ async Task Handler(SQSEvent ev, ILambdaContext context) using var logFactory = new LambdaLoggerFactory(context.Logger); var metrics = new ReconcileMetrics(); var scrubber = new ChangelogContentScrubber(logFactory, allowRepos); - var reconciler = new RegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); + var reconciler = new BundleRegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); var shallowReconciler = new ShallowRegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); var processor = new ScrubberProcessor(logFactory, s3Client, publicBucketName, scrubber, reconciler, shallowReconciler, metrics); diff --git a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs b/src/services/Elastic.Changelog/Reconciliation/BundleRegistryReconciler.cs similarity index 99% rename from src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs rename to src/services/Elastic.Changelog/Reconciliation/BundleRegistryReconciler.cs index 05a3d6a89f..526854f54f 100644 --- a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs +++ b/src/services/Elastic.Changelog/Reconciliation/BundleRegistryReconciler.cs @@ -51,7 +51,7 @@ public sealed class ReconcileConflictException(string message) : Exception(messa /// discovery starts from PR lists, so those manifests stay client-authored pass-through until /// Phase 3 retires them entirely. /// -public sealed class RegistryReconciler( +public sealed class BundleRegistryReconciler( ILoggerFactory logFactory, IAmazonS3 s3Client, string publicBucketName, @@ -73,7 +73,7 @@ public sealed class RegistryReconciler( // First-heal of a large group GETs every unlisted YAML; keep those reads bounded. private const int MaxParallelReads = 4; - private readonly ILogger _logger = logFactory.CreateLogger(); + private readonly ILogger _logger = logFactory.CreateLogger(); private readonly TimeProvider _time = timeProvider ?? TimeProvider.System; private readonly TimeSpan _retryBaseDelay = retryBaseDelay ?? TimeSpan.FromMilliseconds(200); private readonly ReconcileMetrics _metrics = metrics ?? new ReconcileMetrics(); diff --git a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs index 0de6ee1072..bef62e55b7 100644 --- a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs +++ b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs @@ -29,7 +29,7 @@ public sealed class ScrubberProcessor( IAmazonS3 s3Client, string publicBucketName, IChangelogContentScrubber scrubber, - RegistryReconciler reconciler, + BundleRegistryReconciler reconciler, ShallowRegistryReconciler shallowReconciler, ReconcileMetrics? metrics = null ) diff --git a/src/services/Elastic.Changelog/Uploading/Registry.cs b/src/services/Elastic.Changelog/Uploading/Registry.cs index ff81cb0b4c..17d16f3ee3 100644 --- a/src/services/Elastic.Changelog/Uploading/Registry.cs +++ b/src/services/Elastic.Changelog/Uploading/Registry.cs @@ -13,8 +13,11 @@ namespace Elastic.Changelog.Uploading; /// /// /// Stored at bundle/{product}/registry.json (bundle index) or -/// changelog/{org}/{repo}/{branch}/registry.json (changelog-entry index) in the changelog bundles -/// bucket. The scrubber Lambda mirrors it verbatim to the public bucket (pass-through). +/// changelog/{org}/{repo}/{branch}/registry.json (changelog-entry index). Ownership differs +/// per tree: public bundle indexes are produced by the scrubber Lambda's +/// from public-bucket state, while +/// changelog-entry indexes remain client-authored and are mirrored verbatim to the public bucket +/// (pass-through) until Phase 3 of elastic/docs-eng-team#688 retires them. /// public sealed record Registry { @@ -72,16 +75,19 @@ public sealed record RegistryBundle public string? Target { get; init; } /// - /// S3 ETag of the bundle object as uploaded to the private bundles bucket (pre-scrub). - /// For single-part uploads smaller than - /// this is the MD5 of the body. + /// S3 ETag of the object in the bucket this manifest describes. For single-part uploads smaller + /// than this is + /// the MD5 of the body. /// /// - /// Best-effort identity / change hint only. The public (CDN) object is produced by the changelog - /// scrubber Lambda, which rewrites any bundle that contains private references — so for scrubbed - /// bundles this value will not match the public object's ETag. Consumers MUST NOT use it - /// for integrity checks or HTTP cache validation against the public bucket; use the CDN response's - /// own ETag for that. It is safe to use to detect whether a bundle changed between manifest reads. + /// Whose ETag this is follows the manifest's producer. Bundle indexes written by the scrubber + /// Lambda's record the public + /// (post-scrub) object's ETag, valid for HTTP cache validation against the CDN. Client-authored + /// manifests — changelog-entry indexes, and everything in the private bucket — record the + /// private (pre-scrub) upload's ETag: a best-effort change hint that will not + /// match the public object for scrubbed content, so consumers MUST NOT use it for integrity + /// checks or cache validation there. Either way it is safe for detecting that an entry changed + /// between manifest reads. /// public required string ETag { get; init; } } diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/BundleRegistryReconcilerTests.cs similarity index 96% rename from tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs rename to tests/Elastic.Changelog.Tests/Reconciliation/BundleRegistryReconcilerTests.cs index 90c90c39c2..70c9851f00 100644 --- a/tests/Elastic.Changelog.Tests/Reconciliation/RegistryReconcilerTests.cs +++ b/tests/Elastic.Changelog.Tests/Reconciliation/BundleRegistryReconcilerTests.cs @@ -10,18 +10,18 @@ namespace Elastic.Changelog.Tests.Reconciliation; -public class RegistryReconcilerTests +public class BundleRegistryReconcilerTests { private const string PublicBucket = "public-bucket"; private static readonly DateTimeOffset FixedNow = new(2026, 7, 28, 12, 0, 0, TimeSpan.Zero); private readonly FakeS3 _s3 = new(PublicBucket); - private readonly RegistryReconciler _reconciler; + private readonly BundleRegistryReconciler _reconciler; private readonly ReconcileMetrics _metrics = new(); - public RegistryReconcilerTests() => - _reconciler = new RegistryReconciler( + public BundleRegistryReconcilerTests() => + _reconciler = new BundleRegistryReconciler( NullLoggerFactory.Instance, _s3.Client, PublicBucket, @@ -68,7 +68,7 @@ private string SeedBundle(ChangelogScope scope, string file, string target, stri _s3.Seed(PublicBucket, scope.Prefix + file, BundleYaml(product ?? scope.Group, target)); private void SeedManifest(ChangelogScope scope, params RegistryBundle[] bundles) => - SeedManifest(scope, RegistryReconciler.Producer, Registry.CurrentSchemaVersion, bundles); + SeedManifest(scope, BundleRegistryReconciler.Producer, Registry.CurrentSchemaVersion, bundles); private void SeedManifest(ChangelogScope scope, string? producer, int schemaVersion, params RegistryBundle[] bundles) { @@ -112,7 +112,7 @@ public async Task ReconcileGroup_HealsEntriesMissingFromManifest() manifest.Bundles.Select(b => b.File).Should().Equal( "es-9.4.0.yaml", "es-9.3.0.yaml", "es-9.2.0.yaml", "es-9.1.0.yaml"); manifest.Bundles.Should().OnlyContain(b => b.Target != null); - manifest.Producer.Should().Be(RegistryReconciler.Producer); + manifest.Producer.Should().Be(BundleRegistryReconciler.Producer); // 1 and 2 were ETag-reused: only the manifest itself plus 3 and 4 were read. _s3.GetsFor(PublicBucket).Should().BeEquivalentTo([ @@ -248,7 +248,7 @@ public async Task ReconcileGroup_ProducerMismatch_RecomputesEverythingAndWritesE outcome.Should().Be(GroupReconcileOutcome.Written); _s3.GetsFor(PublicBucket).Should().Contain(scope.Prefix + "es-9.1.0.yaml", "producer mismatch disables ETag reuse"); var manifest = WrittenManifest(); - manifest.Producer.Should().Be(RegistryReconciler.Producer); + manifest.Producer.Should().Be(BundleRegistryReconciler.Producer); manifest.Bundles.Single().Should().BeEquivalentTo(new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); } @@ -412,7 +412,7 @@ public async Task ReconcileGroup_NewerSchemaManifest_IsReportedAndLeftUntouched( { var scope = BundleScope(); _ = SeedBundle(scope, "es-9.1.0.yaml", "9.1.0"); - SeedManifest(scope, RegistryReconciler.Producer, schemaVersion: Registry.CurrentSchemaVersion + 1); + SeedManifest(scope, BundleRegistryReconciler.Producer, schemaVersion: Registry.CurrentSchemaVersion + 1); var before = _s3.ContentOf(PublicBucket, scope.RegistryKey); var outcome = await _reconciler.ReconcileGroupAsync(scope, Ctx); diff --git a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs index e0a10cb2a4..7ae9d72f49 100644 --- a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs +++ b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs @@ -30,7 +30,7 @@ public ScrubberProcessorTests() _ = A.CallTo(() => _scrubber.ScrubAsync(A._, A._, A._)) .ReturnsLazily((string _, string content, Cancel _) => Task.FromResult("scrubbed: " + content)); - var reconciler = new RegistryReconciler( + var reconciler = new BundleRegistryReconciler( NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); var shallowReconciler = new ShallowRegistryReconciler( NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); @@ -116,7 +116,7 @@ public async Task Process_BundleRegistryKeyEvents_NeverCopyOrDelete_OnlyTriggerA failed.Should().BeEmpty(); // The public manifest was reconciled from the listing — not copied from the private one. var manifest = PublicManifest("bundle/elasticsearch/registry.json"); - manifest.Producer.Should().Be(RegistryReconciler.Producer); + manifest.Producer.Should().Be(BundleRegistryReconciler.Producer); manifest.Bundles.Select(b => b.File).Should().Equal("es-9.1.0.yaml"); _s3.GetsFor(PrivateBucket).Should().BeEmpty("the private bundle registry content must never be read for pass-through"); }