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..7b98796bd5 --- /dev/null +++ b/src/infra/docs-lambda-changelog-scrubber/EmfMetricsEmitter.cs @@ -0,0 +1,112 @@ +// 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 System.Text.Json.Serialization; +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). 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 IReadOnlyList MetricDefinitions = + [ + 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) + { + var payload = new EmfPayload + { + Aws = new EmfEnvelope + { + 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)); + } +} + +/// 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; } + + 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; } +} + +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/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..6edd17f89a 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,21 @@ 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 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); - 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/BundleRegistryReconciler.cs b/src/services/Elastic.Changelog/Reconciliation/BundleRegistryReconciler.cs new file mode 100644 index 0000000000..526854f54f --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/BundleRegistryReconciler.cs @@ -0,0 +1,435 @@ +// 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 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 BundleRegistryReconciler( + 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) + { + 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++) + { + 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) + { + 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/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..02c7aec72f --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ReconcileMetrics.cs @@ -0,0 +1,76 @@ +// 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 _shallowRegistryWrites; + private int _shallowRegistryUnchanged; + 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; + + /// 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; + + 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 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/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/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..bef62e55b7 --- /dev/null +++ b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs @@ -0,0 +1,354 @@ +// 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; 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( + ILoggerFactory logFactory, + IAmazonS3 s3Client, + string publicBucketName, + IChangelogContentScrubber scrubber, + BundleRegistryReconciler reconciler, + ShallowRegistryReconciler shallowReconciler, + 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, 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)]; + } + + private sealed class GroupWork(ChangelogScope scope) + { + public ChangelogScope Scope { get; } = 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 + /// 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 shallowWork = new Dictionary(); + 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, shallowWork); + } + } + 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, work.PassThrough, 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); + } + } + + foreach (var work in shallowWork.Values) + { + ctx.ThrowIfCancellationRequested(); + try + { + await shallowReconciler.ReconcileAsync(work.Kind, work.Scopes.Values, ctx); + } + catch (Exception e) when (e is not OperationCanceledException) + { + _logger.LogError(e, "Shallow map reconcile for the {Kind} tree failed; failing its {Count} contributing message(s)", work.Kind, 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, + Dictionary shallowWork) + { + var hasScope = ChangelogScope.TryFromKey(key, out var scope); + + if (ChangelogKeys.IsRegistry(key)) + { + 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; + } + + 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; + } + + 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, passThrough); + objectWork[key] = work; + } + work.SourceBucket = sourceBucket; + _ = work.MessageIds.Add(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); + } + + 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 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, bool passThrough, Cancel ctx) + { + _metrics.IncrementObjectReconciles(); + + for (var attempt = 1; attempt <= MaxObjectAttempts; attempt++) + { + ctx.ThrowIfCancellationRequested(); + + var snapshot = await TryGetPrivateObject(sourceBucket, key, ctx); + if (snapshot is { } source) + { + 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 + { + 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, string contentType, Cancel ctx) => + _ = await s3Client.PutObjectAsync(new PutObjectRequest + { + BucketName = publicBucketName, + Key = key, + ContentBody = content, + ContentType = contentType + }, 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..17d16f3ee3 100644 --- a/src/services/Elastic.Changelog/Uploading/Registry.cs +++ b/src/services/Elastic.Changelog/Uploading/Registry.cs @@ -13,15 +13,29 @@ 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 { + /// 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 @@ -61,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/BundleRegistryReconcilerTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/BundleRegistryReconcilerTests.cs new file mode 100644 index 0000000000..70c9851f00 --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Reconciliation/BundleRegistryReconcilerTests.cs @@ -0,0 +1,443 @@ +// 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 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 BundleRegistryReconciler _reconciler; + private readonly ReconcileMetrics _metrics = new(); + + public BundleRegistryReconcilerTests() => + _reconciler = new BundleRegistryReconciler( + 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, BundleRegistryReconciler.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(BundleRegistryReconciler.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(BundleRegistryReconciler.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_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"); + + var act = async () => await _reconciler.ReconcileGroupAsync(scope, Ctx); + + _ = await act.Should().ThrowAsync(); + _s3.Puts.Should().BeEmpty(); + } + + [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, BundleRegistryReconciler.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/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/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 new file mode 100644 index 0000000000..7ae9d72f49 --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs @@ -0,0 +1,370 @@ +// 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 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); + _processor = new ScrubberProcessor( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, shallowReconciler, _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)!; + + private SortedDictionary ShallowMap(string mapKey) => + JsonSerializer.Deserialize(_s3.ContentOf(PublicBucket, mapKey), ShallowRegistryJsonContext.Default.SortedDictionaryStringString)!; + + [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_BundleRegistryKeyEvents_NeverCopyOrDelete_OnlyTriggerAGroupReconcile() + { + // 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()); + + 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(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"); + } + + [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] + 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 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", "{}"); + + 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 + """; +}