diff --git a/docs/cli-schema.json b/docs/cli-schema.json index ef677c35be..63b62a4dd7 100644 --- a/docs/cli-schema.json +++ b/docs/cli-schema.json @@ -4433,7 +4433,230 @@ ] } ], - "namespaces": [] + "namespaces": [ + { + "segment": "registry", + "summary": "Operate on the scrubber-owned public changelog registries.", + "options": [], + "commands": [ + { + "path": [ + "changelog", + "registry" + ], + "name": "reconcile", + "summary": "Send explicit reconcile messages to the scrubber queue so the Lambda performs a full group heal (object-level reconcile over the union of both buckets, then a registry rebuild from public state) for every planned group. This command never mutates S3 itself \u2014 the scrubber Lambda stays the public bucket\u0027s single writer. Convergent: re-running re-plans against current state. Enqueuing is not reconciling \u2014 gate on \u0060changelog registry verify\u0060 after the queue drains and the DLQ is empty.", + "usage": "docs-builder changelog registry reconcile --s3-bucket-name \u003Cstring\u003E --public-s3-bucket-name \u003Cstring\u003E --queue-url \u003Cstring\u003E [options]", + "examples": [], + "parameters": [ + { + "role": "flag", + "name": "s3-bucket-name", + "type": "string", + "required": true, + "summary": "Private changelog bundles bucket to plan from." + }, + { + "role": "flag", + "name": "public-s3-bucket-name", + "type": "string", + "required": true, + "summary": "Public (CDN) changelog bundles bucket to plan from." + }, + { + "role": "flag", + "name": "queue-url", + "type": "string", + "required": true, + "summary": "URL of the scrubber SQS queue to send reconcile messages to." + }, + { + "role": "flag", + "name": "product", + "type": "string", + "required": false, + "summary": "Only reconcile this bundle group (bundle/{product}/). Mutually exclusive with --owner/--repo/--branch." + }, + { + "role": "flag", + "name": "owner", + "type": "string", + "required": false, + "summary": "GitHub owner of a single changelog-pool group to reconcile." + }, + { + "role": "flag", + "name": "repo", + "type": "string", + "required": false, + "summary": "Repository of a single changelog-pool group to reconcile." + }, + { + "role": "flag", + "name": "branch", + "type": "string", + "required": false, + "summary": "Branch of a single changelog-pool group to reconcile (verbatim; slashes allowed)." + }, + { + "role": "dryRun", + "name": "dry-run", + "type": "boolean", + "required": false, + "summary": "Print the group plan without sending anything.", + "defaultValue": "false" + }, + { + "role": "flag", + "name": "yes", + "type": "boolean", + "required": false, + "summary": "Skip the interactive confirmation (required when stdin is not a terminal).", + "defaultValue": "false" + }, + { + "role": "flag", + "name": "log-level", + "shortName": "l", + "type": "enum", + "required": false, + "summary": "Minimum log level. Default: information", + "enumValues": [ + "trace", + "debug", + "information", + "warning", + "error", + "critical", + "none" + ] + }, + { + "role": "flag", + "name": "config-source", + "shortName": "c", + "type": "enum", + "required": false, + "summary": "Override the configuration source: local, remote", + "enumValues": [ + "local", + "remote", + "embedded" + ] + }, + { + "role": "flag", + "name": "skip-private-repositories", + "type": "boolean", + "required": false, + "summary": "Skip cloning private repositories" + } + ], + "intent": { + "destructive": true, + "idempotent": true, + "scope": "global", + "requiresAuth": true + } + }, + { + "path": [ + "changelog", + "registry" + ], + "name": "verify", + "summary": "Compare each planned group\u0027s public registry.json against what a reconcile of the current public listing would write, and report divergence (missing, stale, corrupt, object-divergent; unsupported schema reported distinctly). Strictly read-only. Zero divergence across the plan is the cutover completion gate \u2014 and the standing drift-diagnosis tool afterwards.", + "usage": "docs-builder changelog registry verify --s3-bucket-name \u003Cstring\u003E --public-s3-bucket-name \u003Cstring\u003E [options]", + "examples": [], + "parameters": [ + { + "role": "flag", + "name": "s3-bucket-name", + "type": "string", + "required": true, + "summary": "Private changelog bundles bucket to plan from." + }, + { + "role": "flag", + "name": "public-s3-bucket-name", + "type": "string", + "required": true, + "summary": "Public (CDN) changelog bundles bucket to verify." + }, + { + "role": "flag", + "name": "product", + "type": "string", + "required": false, + "summary": "Only verify this bundle group (bundle/{product}/). Mutually exclusive with --owner/--repo/--branch." + }, + { + "role": "flag", + "name": "owner", + "type": "string", + "required": false, + "summary": "GitHub owner of a single changelog-pool group to verify." + }, + { + "role": "flag", + "name": "repo", + "type": "string", + "required": false, + "summary": "Repository of a single changelog-pool group to verify." + }, + { + "role": "flag", + "name": "branch", + "type": "string", + "required": false, + "summary": "Branch of a single changelog-pool group to verify (verbatim; slashes allowed)." + }, + { + "role": "flag", + "name": "log-level", + "shortName": "l", + "type": "enum", + "required": false, + "summary": "Minimum log level. Default: information", + "enumValues": [ + "trace", + "debug", + "information", + "warning", + "error", + "critical", + "none" + ] + }, + { + "role": "flag", + "name": "config-source", + "shortName": "c", + "type": "enum", + "required": false, + "summary": "Override the configuration source: local, remote", + "enumValues": [ + "local", + "remote", + "embedded" + ] + }, + { + "role": "flag", + "name": "skip-private-repositories", + "type": "boolean", + "required": false, + "summary": "Skip cloning private repositories" + } + ], + "intent": { + "requiresAuth": true + } + } + ], + "namespaces": [] + } + ] }, { "segment": "codex", diff --git a/docs/cli/changelog/registry/cmd-reconcile.md b/docs/cli/changelog/registry/cmd-reconcile.md new file mode 100644 index 0000000000..3c24f9867c --- /dev/null +++ b/docs/cli/changelog/registry/cmd-reconcile.md @@ -0,0 +1,31 @@ +## Description + +Sends one explicit reconcile message per planned group to the scrubber queue. For each message the Lambda performs a **full group heal**: an object-level reconcile over the union of both buckets' listings (scrub and copy whatever is live in the private bucket, delete public objects nothing backs), then a rebuild of the group's public `registry.json` from the public listing. This recovers drift that no pending S3 event would ever repair — lost or DLQ-expired scrub events, ad-hoc uploads, orphaned public objects. + +Without a scope filter, the plan is the union of groups discovered in **both** buckets, so orphan public groups (including groups that only have a leftover manifest) are covered. Use `--product`, or `--owner`/`--repo`/`--branch` together, to reconcile a single group. + +The command is convergent: re-running it re-plans against current state and the Lambda's writes are conditional, so overlapping runs cannot corrupt a manifest. It is also deliberately indirect — the CLI only sends queue messages, keeping the scrubber Lambda the sole writer of the public bucket. + +Every run stamps one correlation id on all its messages and prints a ledger line per group (`group`, SQS `message-id`, `correlation-id`). **Enqueuing is not reconciling**: after a run, watch the queue drain (oldest-message-age ≈ 0), triage anything that reaches the DLQ, and gate on [](/cli/changelog/registry/verify.md) reporting zero divergence. + +## Examples + +Preview the full plan without sending anything: + +```bash +docs-builder changelog registry reconcile \ + --s3-bucket-name elastic-docs-v3-changelog-bundles-private \ + --public-s3-bucket-name elastic-docs-v3-changelog-bundles \ + --queue-url https://sqs.us-east-1.amazonaws.com//elastic-docs-v3-changelog-scrub-queue \ + --dry-run +``` + +Reconcile a single product's bundle registry non-interactively: + +```bash +docs-builder changelog registry reconcile \ + --s3-bucket-name elastic-docs-v3-changelog-bundles-private \ + --public-s3-bucket-name elastic-docs-v3-changelog-bundles \ + --queue-url https://sqs.us-east-1.amazonaws.com//elastic-docs-v3-changelog-scrub-queue \ + --product elasticsearch --yes +``` diff --git a/docs/cli/changelog/registry/cmd-verify.md b/docs/cli/changelog/registry/cmd-verify.md new file mode 100644 index 0000000000..135ce935d8 --- /dev/null +++ b/docs/cli/changelog/registry/cmd-verify.md @@ -0,0 +1,21 @@ +## Description + +Read-only drift diagnosis: for every planned group, compares the public `registry.json` against what a reconcile of the current public listing would write — the exact same listing spec and entry rules the scrubber Lambda uses — and reports each divergence: + +| Kind | Meaning | +|---|---| +| `Missing` | A public object (or the manifest itself) the registry should describe but doesn't. | +| `Stale` | A manifest entry (or whole manifest) describing something no longer in the bucket, or manifest metadata a reconcile would rewrite. | +| `Corrupt` | The manifest exists but cannot be parsed. | +| `ObjectDivergent` | File present on both sides, but the recorded ETag or target disagrees with the object. | +| `UnsupportedSchema` | The manifest declares a newer `schema_version` than this tool understands. Reported distinctly and never rewritten. | + +The command exits non-zero when any group diverges. Zero divergence across the plan is the completion gate after a [](/cli/changelog/registry/reconcile.md) run — and the standing way to answer "is the registry trustworthy right now?". + +## Examples + +```bash +docs-builder changelog registry verify \ + --s3-bucket-name elastic-docs-v3-changelog-bundles-private \ + --public-s3-bucket-name elastic-docs-v3-changelog-bundles +``` diff --git a/docs/cli/changelog/registry/index.md b/docs/cli/changelog/registry/index.md new file mode 100644 index 0000000000..a37368f811 --- /dev/null +++ b/docs/cli/changelog/registry/index.md @@ -0,0 +1,3 @@ +Operate on the public changelog bundle registries owned by the scrubber Lambda. + +Each group of published changelog artifacts — `bundle/{product}/` or `changelog/{org}/{repo}/{branch}/` — carries a `registry.json` manifest in the **public** bucket, produced exclusively by the scrubber Lambda from the bucket's actual state. These commands are the operator surface for that machinery: `reconcile` asks the Lambda to converge groups (via explicit queue messages — the CLI never writes to S3 itself), and `verify` reports, read-only, whether each public manifest matches its public listing. diff --git a/src/infra/docs-lambda-changelog-scrubber/Program.cs b/src/infra/docs-lambda-changelog-scrubber/Program.cs index c1f668129c..d87eb63639 100644 --- a/src/infra/docs-lambda-changelog-scrubber/Program.cs +++ b/src/infra/docs-lambda-changelog-scrubber/Program.cs @@ -17,6 +17,10 @@ var publicBucketName = Environment.GetEnvironmentVariable("PUBLIC_BUCKET_NAME") ?? throw new InvalidOperationException("PUBLIC_BUCKET_NAME environment variable is required"); +// Optional: only explicit reconcile messages (full group heals) need to list the private bucket +// by name — S3 events carry their source bucket. Without it, reconcile messages are rejected. +var privateBucketName = Environment.GetEnvironmentVariable("PRIVATE_BUCKET_NAME"); + var allowRepos = BuildAllowlist(); await LambdaBootstrapBuilder @@ -55,7 +59,7 @@ async Task Handler(SQSEvent ev, ILambdaContext context) var metrics = new ReconcileMetrics(); var scrubber = new ChangelogContentScrubber(logFactory, allowRepos); var reconciler = new RegistryReconciler(logFactory, s3Client, publicBucketName, metrics: metrics); - var processor = new ScrubberProcessor(logFactory, s3Client, publicBucketName, scrubber, reconciler, metrics); + var processor = new ScrubberProcessor(logFactory, s3Client, publicBucketName, scrubber, reconciler, metrics, privateBucketName); var messages = ev.Records.Select(r => new ScrubberQueueMessage(r.MessageId, r.Body)).ToList(); var failedIds = await processor.ProcessAsync(messages, CancellationToken.None); diff --git a/src/services/Elastic.Changelog/Elastic.Changelog.csproj b/src/services/Elastic.Changelog/Elastic.Changelog.csproj index db7c27cc6f..6c89c81174 100644 --- a/src/services/Elastic.Changelog/Elastic.Changelog.csproj +++ b/src/services/Elastic.Changelog/Elastic.Changelog.csproj @@ -17,6 +17,7 @@ + diff --git a/src/services/Elastic.Changelog/Reconciliation/ChangelogGroupDiscovery.cs b/src/services/Elastic.Changelog/Reconciliation/ChangelogGroupDiscovery.cs new file mode 100644 index 0000000000..27b6c707a9 --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ChangelogGroupDiscovery.cs @@ -0,0 +1,66 @@ +// Licensed to Elasticsearch B.V under one or more agreements. +// Elasticsearch B.V licenses this file to you under the Apache 2.0 License. +// See the LICENSE file in the project root for more information + +using Amazon.S3; +using Amazon.S3.Model; +using Elastic.Documentation.Configuration.ReleaseNotes; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// Enumerates every registry group present in a bucket by walking the bundle/ and +/// changelog/ prefixes and deriving each key's scope. Registry keys count too, so a group +/// that only has an orphaned manifest left is still planned (its reconcile deletes the manifest). +/// The reconcile/verify planners union this across both buckets so orphan public groups are +/// covered as well. +/// +public static class ChangelogGroupDiscovery +{ + /// Every scope with at least one key in , keyed by prefix. + public static async Task> DiscoverGroupsAsync( + IAmazonS3 s3Client, + string bucketName, + Cancel ctx) + { + var scopes = new Dictionary(StringComparer.Ordinal); + foreach (var prefix in new[] { ChangelogKeys.BundlePrefix, ChangelogKeys.ChangelogPrefix }) + { + var request = new ListObjectsV2Request + { + BucketName = bucketName, + Prefix = prefix + }; + + ListObjectsV2Response response; + do + { + response = await s3Client.ListObjectsV2Async(request, ctx); + foreach (var obj in response.S3Objects ?? []) + { + if (ChangelogScope.TryFromKey(obj.Key, out var scope)) + _ = scopes.TryAdd(scope.Prefix, scope); + } + request.ContinuationToken = response.NextContinuationToken; + } while (response.IsTruncated == true); + } + + return scopes; + } + + /// The union of both buckets' groups, ordered by prefix for a stable plan. + public static async Task> DiscoverUnionAsync( + IAmazonS3 s3Client, + string privateBucketName, + string publicBucketName, + Cancel ctx) + { + var union = new Dictionary(StringComparer.Ordinal); + foreach (var (prefix, scope) in await DiscoverGroupsAsync(s3Client, privateBucketName, ctx)) + _ = union.TryAdd(prefix, scope); + foreach (var (prefix, scope) in await DiscoverGroupsAsync(s3Client, publicBucketName, ctx)) + _ = union.TryAdd(prefix, scope); + + return [.. union.Values.OrderBy(s => s.Prefix, StringComparer.Ordinal)]; + } +} diff --git a/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryArguments.cs b/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryArguments.cs new file mode 100644 index 0000000000..9fa9cb4683 --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryArguments.cs @@ -0,0 +1,88 @@ +// Licensed to Elasticsearch B.V under one or more agreements. +// Elasticsearch B.V licenses this file to you under the Apache 2.0 License. +// See the LICENSE file in the project root for more information + +using Elastic.Documentation.Diagnostics; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// Scope selection shared by the registry commands: with no filter the plan covers every group +/// discovered in the union of both buckets; a bundle scope is addressed by , +/// a changelog-pool scope by //. +/// +public record ChangelogRegistryScopeArguments +{ + /// Product of a bundle scope (bundle/{product}/). Mutually exclusive with the owner/repo/branch form. + public string? Product { get; init; } + + /// GitHub owner of a changelog-pool scope (changelog/{org}/{repo}/{branch}/). + public string? Owner { get; init; } + + /// Repository of a changelog-pool scope. + public string? Repo { get; init; } + + /// Branch of a changelog-pool scope (verbatim; slashes become key segments). + public string? Branch { get; init; } + + /// The private changelog bundles bucket. + public required string S3BucketName { get; init; } + + /// The scrubber-owned public bucket. + public required string PublicS3BucketName { get; init; } + + /// + /// Resolves the optional scope filter. True with a null when no + /// filter was given (plan every discovered group); false with an error when both forms are + /// mixed or a segment fails validation. + /// + public bool TryResolveScopeFilter(IDiagnosticsCollector collector, out ChangelogScope? scope) + { + scope = null; + var hasProduct = !string.IsNullOrWhiteSpace(Product); + var hasPool = !string.IsNullOrWhiteSpace(Owner) || !string.IsNullOrWhiteSpace(Repo) || !string.IsNullOrWhiteSpace(Branch); + + if (!hasProduct && !hasPool) + return true; + + if (hasProduct && hasPool) + { + collector.EmitError(string.Empty, + "Specify at most one scope: --product for a bundle scope, or --owner, --repo, and --branch together for a changelog scope."); + return false; + } + + if (hasProduct) + { + if (ChangelogScope.TryCreateBundle(Product, out scope)) + return true; + + collector.EmitError(string.Empty, $"Invalid product \"{Product}\" (must match [a-zA-Z0-9_-]+)."); + return false; + } + + if (ChangelogScope.TryCreateChangelog(Owner, Repo, Branch, out scope)) + return true; + + collector.EmitError(string.Empty, + $"Invalid changelog scope \"{Owner ?? ""}/{Repo ?? ""}/{Branch ?? ""}\": " + + "--owner, --repo, and --branch are all required and each segment must be a valid key segment."); + return false; + } +} + +/// Arguments for . +public sealed record ChangelogRegistryReconcileArguments : ChangelogRegistryScopeArguments +{ + /// URL of the scrubber SQS queue the reconcile messages are sent to. + public required string QueueUrl { get; init; } + + /// Print the group plan without sending anything. + public bool DryRun { get; init; } + + /// Skip the interactive confirmation (required in non-interactive contexts). + public bool AssumeYes { get; init; } +} + +/// Arguments for . +public sealed record ChangelogRegistryVerifyArguments : ChangelogRegistryScopeArguments; diff --git a/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryReconcileService.cs b/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryReconcileService.cs new file mode 100644 index 0000000000..722f0e60ab --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryReconcileService.cs @@ -0,0 +1,108 @@ +// 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.S3; +using Amazon.SQS; +using Amazon.SQS.Model; +using Elastic.Documentation.Diagnostics; +using Elastic.Documentation.Services; +using Microsoft.Extensions.Logging; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// The cutover/heal entry point of elastic/docs-eng-team#688 Phase 2: plans the affected groups +/// (one scope, or the union of both buckets' groups so orphan public groups are covered) and +/// sends one explicit, versioned reconcile message per group to the scrubber queue. The Lambda +/// does the actual work — this command never mutates S3 itself, which keeps the single-writer +/// invariant intact. Convergent by design: re-running re-plans against current state. +/// +public sealed class ChangelogRegistryReconcileService( + ILoggerFactory logFactory, + IAmazonS3? s3Client = null, + IAmazonSQS? sqsClient = null, + Func? confirmationReader = null +) : IService +{ + private readonly ILogger _logger = logFactory.CreateLogger(); + + public async Task Reconcile(IDiagnosticsCollector collector, ChangelogRegistryReconcileArguments args, Cancel ctx) + { + if (!args.TryResolveScopeFilter(collector, out var filter)) + return false; + + using var defaultS3 = s3Client is null ? new AmazonS3Client() : null; + var s3 = s3Client ?? defaultS3!; + + var plan = filter is not null + ? [filter] + : await ChangelogGroupDiscovery.DiscoverUnionAsync(s3, args.S3BucketName, args.PublicS3BucketName, ctx); + + if (plan.Count == 0) + { + _logger.LogInformation("No registry groups found in {Private} or {Public}; nothing to reconcile", args.S3BucketName, args.PublicS3BucketName); + return true; + } + + _logger.LogInformation("Reconcile plan: {Count} group(s)", plan.Count); + foreach (var scope in plan) + _logger.LogInformation(" {Kind,-9} {Group}", scope.Kind == ChangelogScopeKind.Bundle ? "bundle" : "changelog", scope.Group); + + if (args.DryRun) + { + _logger.LogInformation("[dry-run] Would send {Count} reconcile message(s) to {QueueUrl}", plan.Count, args.QueueUrl); + return true; + } + + if (!args.AssumeYes && !Confirm(plan.Count, collector)) + return false; + + // One correlation id per run ties every ledger line to this invocation; the cutover + // checkpoint replays this ledger through `registry verify`. + var correlationId = Guid.NewGuid().ToString("N"); + using var defaultSqs = sqsClient is null ? new AmazonSQSClient() : null; + var sqs = sqsClient ?? defaultSqs!; + + var sent = 0; + foreach (var scope in plan) + { + ctx.ThrowIfCancellationRequested(); + + var message = ReconcileQueueMessage.For(scope, correlationId); + var response = await sqs.SendMessageAsync(new SendMessageRequest + { + QueueUrl = args.QueueUrl, + MessageBody = message.ToJson() + }, ctx); + sent++; + _logger.LogInformation( + "ledger: group={Group} scope={Scope} message-id={MessageId} correlation-id={CorrelationId}", + scope.Group, message.Scope, response.MessageId, correlationId); + } + + _logger.LogInformation( + "Sent {Sent} reconcile message(s) (correlation-id {CorrelationId}). Enqueuing is not reconciling: " + + "watch the queue drain, triage any DLQ entry, then gate on `changelog registry verify`.", + sent, correlationId); + return true; + } + + private bool Confirm(int groupCount, IDiagnosticsCollector collector) + { + if (confirmationReader is null && Console.IsInputRedirected) + { + collector.EmitError(string.Empty, + "Refusing to send reconcile messages without confirmation in a non-interactive session; re-run with --yes (or --dry-run to preview)."); + return false; + } + + Console.Write($"Send reconcile messages for {groupCount} group(s)? Every public registry in the plan will be rewritten end-to-end. [y/N] "); + var answer = confirmationReader is not null ? confirmationReader() : Console.ReadLine(); + if (string.Equals(answer?.Trim(), "y", StringComparison.OrdinalIgnoreCase)) + return true; + + collector.EmitError(string.Empty, "Aborted: confirmation declined."); + return false; + } +} diff --git a/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryVerifyService.cs b/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryVerifyService.cs new file mode 100644 index 0000000000..1c26b319e6 --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ChangelogRegistryVerifyService.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 + +using Amazon.S3; +using Elastic.Documentation.Diagnostics; +using Elastic.Documentation.Services; +using Microsoft.Extensions.Logging; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// The read-only sibling of changelog registry reconcile and the standing drift-diagnosis +/// tool: for each planned group, compares the public manifest against what a reconcile of the +/// current public listing would write — same listing spec and entry rules by construction +/// () — and reports divergence. Zero divergence +/// across the plan is the cutover completion gate of elastic/docs-eng-team#688. +/// +public sealed class ChangelogRegistryVerifyService( + ILoggerFactory logFactory, + IAmazonS3? s3Client = null +) : IService +{ + private readonly ILogger _logger = logFactory.CreateLogger(); + + public async Task Verify(IDiagnosticsCollector collector, ChangelogRegistryVerifyArguments args, Cancel ctx) + { + if (!args.TryResolveScopeFilter(collector, out var filter)) + return false; + + using var defaultS3 = s3Client is null ? new AmazonS3Client() : null; + var s3 = s3Client ?? defaultS3!; + + var plan = filter is not null + ? [filter] + : await ChangelogGroupDiscovery.DiscoverUnionAsync(s3, args.S3BucketName, args.PublicS3BucketName, ctx); + + if (plan.Count == 0) + { + _logger.LogInformation("No registry groups found in {Private} or {Public}; nothing to verify", args.S3BucketName, args.PublicS3BucketName); + return true; + } + + var reconciler = new RegistryReconciler(logFactory, s3, args.PublicS3BucketName); + var divergentGroups = 0; + var findings = 0; + + foreach (var scope in plan) + { + ctx.ThrowIfCancellationRequested(); + + var divergences = await reconciler.VerifyGroupAsync(scope, ctx); + if (divergences.Count == 0) + { + _logger.LogInformation("{Scope}: converged", scope); + continue; + } + + divergentGroups++; + findings += divergences.Count; + foreach (var divergence in divergences) + collector.EmitWarning(string.Empty, $"{scope.Prefix}{divergence.File}: [{divergence.Kind}] {divergence.Detail}"); + } + + if (divergentGroups > 0) + { + collector.EmitError(string.Empty, + $"{divergentGroups} of {plan.Count} group(s) diverge from their public listing ({findings} finding(s)). " + + "Run `changelog registry reconcile` to converge them, then verify again."); + return false; + } + + _logger.LogInformation("All {Count} group(s) converged: every public manifest matches its public listing", plan.Count); + return true; + } +} diff --git a/src/services/Elastic.Changelog/Reconciliation/ReconcileQueueMessage.cs b/src/services/Elastic.Changelog/Reconciliation/ReconcileQueueMessage.cs new file mode 100644 index 0000000000..600e0a8c6c --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/ReconcileQueueMessage.cs @@ -0,0 +1,122 @@ +// 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; +using System.Text.Json.Serialization; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// The explicit reconcile request `changelog registry reconcile` sends to the scrubber queue +/// (elastic/docs-eng-team#688 Phase 2). A versioned, discriminated envelope: the scrubber treats +/// any body whose kind is reconcile as one of these and performs a full group +/// heal — object-level reconcile over the union of both buckets' group listings, then the +/// group reconcile — which is what makes a lost or DLQ-expired scrub event recoverable by message. +/// +public sealed record ReconcileQueueMessage +{ + /// The discriminator value marking a body as a reconcile request. + public const string ReconcileKind = "reconcile"; + + /// The envelope version this producer writes and the consumer accepts. + public const int CurrentVersion = 1; + + /// Body discriminator; always for this type. + public string? Kind { get; init; } + + /// Envelope version; the consumer rejects anything but . + public int Version { get; init; } + + /// The scope family: bundle or changelog. + public string? Scope { get; init; } + + /// The group inside the scope: a product, or {org}/{repo}/{branch}. + public string? Group { get; init; } + + /// Caller-chosen id tying this message to a cutover ledger entry. + public string? CorrelationId { get; init; } + + /// True when parses as JSON whose kind is — regardless of whether the rest validates. + public static bool TryRead(string body, [NotNullWhen(true)] out ReconcileQueueMessage? message) + { + message = null; + if (!body.Contains("\"kind\"", StringComparison.OrdinalIgnoreCase)) + return false; + try + { + var parsed = JsonSerializer.Deserialize(body, ReconcileQueueMessageJsonContext.Default.ReconcileQueueMessage); + if (parsed is null || !string.Equals(parsed.Kind, ReconcileKind, StringComparison.Ordinal)) + return false; + message = parsed; + return true; + } + catch (JsonException) + { + return false; + } + } + + /// + /// Validates the envelope strictly — version, scope family, and every group segment (via + /// , hence ChangelogKeys) — before anything derives an S3 + /// listing from it. False with a reason when the message must be rejected. + /// + public bool TryResolveScope([NotNullWhen(true)] out ChangelogScope? scope, [NotNullWhen(false)] out string? error) + { + scope = null; + if (Version != CurrentVersion) + { + error = $"unsupported reconcile message version {Version} (supported: {CurrentVersion})"; + return false; + } + + switch (Scope) + { + case "bundle": + if (!ChangelogScope.TryCreateBundle(Group, out scope)) + { + error = $"invalid bundle group \"{Group}\""; + return false; + } + break; + case "changelog": + var segments = Group?.Split('/') ?? []; + if (segments.Length < 3 + || !ChangelogScope.TryCreateChangelog(segments[0], segments[1], string.Join('/', segments[2..]), out scope)) + { + error = $"invalid changelog group \"{Group}\" (expected {{org}}/{{repo}}/{{branch}})"; + return false; + } + break; + default: + error = $"unknown reconcile scope \"{Scope}\""; + return false; + } + + error = null; + return true; + } + + /// Builds the message for one scope, stamped with the caller's correlation id. + public static ReconcileQueueMessage For(ChangelogScope scope, string correlationId) => new() + { + Kind = ReconcileKind, + Version = CurrentVersion, + Scope = scope.Kind == ChangelogScopeKind.Bundle ? "bundle" : "changelog", + Group = scope.Group, + CorrelationId = correlationId + }; + + /// Serializes with the same source-generated context the consumer parses with. + public string ToJson() => + JsonSerializer.Serialize(this, ReconcileQueueMessageJsonContext.Default.ReconcileQueueMessage); +} + +[JsonSourceGenerationOptions( + PropertyNamingPolicy = JsonKnownNamingPolicy.SnakeCaseLower, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull +)] +[JsonSerializable(typeof(ReconcileQueueMessage))] +public sealed partial class ReconcileQueueMessageJsonContext : JsonSerializerContext; diff --git a/src/services/Elastic.Changelog/Reconciliation/RegistryDivergence.cs b/src/services/Elastic.Changelog/Reconciliation/RegistryDivergence.cs new file mode 100644 index 0000000000..092b9b956a --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/RegistryDivergence.cs @@ -0,0 +1,37 @@ +// 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; + +/// How a group's public manifest diverges from its public listing. +public enum RegistryDivergenceKind +{ + /// A public object (or the manifest itself) that the registry should describe but doesn't. + Missing, + + /// A manifest entry (or the whole manifest) describing something no longer in the bucket, or metadata a reconcile would rewrite. + Stale, + + /// The manifest exists but cannot be parsed. + Corrupt, + + /// File present on both sides but the recorded ETag or target disagrees with the object. + ObjectDivergent, + + /// The manifest declares a newer schema than this tool understands; reported distinctly, never rewritten. + UnsupportedSchema +} + +/// One verify finding for a group. +public sealed record RegistryDivergence +{ + /// The divergence family. + public required RegistryDivergenceKind Kind { get; init; } + + /// The file inside the group's prefix (or registry.json for manifest-level findings). + public required string File { get; init; } + + /// Human-readable specifics. + public required string Detail { get; init; } +} diff --git a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs b/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs index a1148ae8f9..e096aeff27 100644 --- a/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs +++ b/src/services/Elastic.Changelog/Reconciliation/RegistryReconciler.cs @@ -155,41 +155,124 @@ public async Task ReconcileGroupAsync(ChangelogScope scop } /// - /// 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. + /// Read-only diagnosis: compares the group's public manifest against what a reconcile of the + /// current public listing would write — same listing spec, same entry rules, zero writes. + /// An empty result means a reconcile would be a no-op for this group. /// - private async Task> ListGroupFiles(ChangelogScope scope, Cancel ctx) + public async Task> VerifyGroupAsync(ChangelogScope scope, Cancel ctx) { - var request = new ListObjectsV2Request + var listing = await ListGroupFiles(scope, ctx); + var existing = await FetchManifest(scope.RegistryKey, ctx); + var divergences = new List(); + + if (listing.Count == 0) { - BucketName = publicBucketName, - Prefix = scope.Prefix, - Delimiter = "/" - }; + if (existing.Exists) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.Stale, + File = ChangelogKeys.RegistryFileName, + Detail = "The group holds no objects but its manifest still exists; a reconcile would delete it (absent ≠ empty for consumers)." + }); + } + return divergences; + } - var files = new List(); - ListObjectsV2Response response; - do + if (!existing.Exists) { - response = await s3Client.ListObjectsV2Async(request, ctx); - foreach (var obj in response.S3Objects ?? []) + divergences.Add(new RegistryDivergence { - var file = obj.Key[scope.Prefix.Length..]; - if (!IsYamlFileName(file) || string.Equals(file, ChangelogKeys.RegistryFileName, StringComparison.Ordinal)) - continue; - files.Add(obj); - _metrics.IncrementObjectsListed(); + Kind = RegistryDivergenceKind.Missing, + File = ChangelogKeys.RegistryFileName, + Detail = $"The group holds {listing.Count} object(s) but no manifest." + }); + return divergences; + } + + if (existing.Corrupt) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.Corrupt, + File = ChangelogKeys.RegistryFileName, + Detail = "The manifest cannot be parsed; a reconcile would rebuild it from the listing." + }); + return divergences; + } + + var manifest = existing.Manifest!; + if (manifest.SchemaVersion > Registry.CurrentSchemaVersion) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.UnsupportedSchema, + File = ChangelogKeys.RegistryFileName, + Detail = $"The manifest declares schema_version {manifest.SchemaVersion} > supported {Registry.CurrentSchemaVersion}; this tool will not touch it." + }); + return divergences; + } + + var trusted = manifest.SchemaVersion == Registry.CurrentSchemaVersion + && string.Equals(manifest.Producer, Producer, StringComparison.Ordinal) + && string.Equals(manifest.Product, scope.Group, StringComparison.Ordinal); + if (!trusted) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.Stale, + File = ChangelogKeys.RegistryFileName, + Detail = $"Manifest metadata is not this producer's (producer \"{manifest.Producer}\", product \"{manifest.Product}\"); a reconcile would rewrite it." + }); + } + + var (desired, _) = await BuildEntries(scope, listing, trusted ? manifest.Bundles : [], ctx); + var desiredByFile = desired.ToDictionary(b => b.File, b => b, StringComparer.Ordinal); + var manifestByFile = manifest.Bundles.ToDictionary(b => b.File, b => b, StringComparer.Ordinal); + + foreach (var (file, entry) in desiredByFile) + { + if (!manifestByFile.TryGetValue(file, out var recorded)) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.Missing, + File = file, + Detail = "The object exists in the public bucket but the manifest does not list it." + }); + } + else if (!string.Equals(recorded.ETag, entry.ETag, StringComparison.Ordinal) + || !string.Equals(recorded.Target, entry.Target, StringComparison.Ordinal)) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.ObjectDivergent, + File = file, + Detail = $"The manifest records (target: {recorded.Target ?? "null"}, etag: {recorded.ETag}) but a reconcile would write (target: {entry.Target ?? "null"}, etag: {entry.ETag})." + }); } - request.ContinuationToken = response.NextContinuationToken; - } while (response.IsTruncated == true); + } - return files; + foreach (var file in manifestByFile.Keys.Where(f => !desiredByFile.ContainsKey(f))) + { + divergences.Add(new RegistryDivergence + { + Kind = RegistryDivergenceKind.Stale, + File = file, + Detail = "The manifest lists an object that is no longer in the public bucket." + }); + } + + return divergences; } - private static bool IsYamlFileName(string file) => - file.Length > 0 - && (file.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase) || file.EndsWith(".yml", StringComparison.OrdinalIgnoreCase)); + private async Task> ListGroupFiles(ChangelogScope scope, Cancel ctx) + { + var files = await S3GroupListing.ListImmediateYamlObjectsAsync(s3Client, publicBucketName, scope, ctx); + for (var i = 0; i < files.Count; i++) + _metrics.IncrementObjectsListed(); + return files; + } private sealed record ManifestState(Registry? Manifest, string? ETag, bool Exists, bool Corrupt); diff --git a/src/services/Elastic.Changelog/Reconciliation/S3GroupListing.cs b/src/services/Elastic.Changelog/Reconciliation/S3GroupListing.cs new file mode 100644 index 0000000000..9706fddef7 --- /dev/null +++ b/src/services/Elastic.Changelog/Reconciliation/S3GroupListing.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.S3; +using Amazon.S3.Model; +using Elastic.Documentation.Configuration.ReleaseNotes; + +namespace Elastic.Changelog.Reconciliation; + +/// +/// The one listing spec every registry operation shares (reconcile, full group heal, verify): +/// a group's objects are the immediate .yaml/.yml children of its prefix. The +/// / delimiter is load-bearing — branches are stored verbatim, so without it +/// changelog/{org}/{repo}/main/ would also sweep in the main/feature/… pool — +/// and pagination runs to completion. The manifest itself and any other non-YAML keys are +/// excluded. +/// +public static class S3GroupListing +{ + /// Lists the group's immediate YAML children in . + public static async Task> ListImmediateYamlObjectsAsync( + IAmazonS3 s3Client, + string bucketName, + ChangelogScope scope, + Cancel ctx) + { + var request = new ListObjectsV2Request + { + BucketName = bucketName, + 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); + } + request.ContinuationToken = response.NextContinuationToken; + } while (response.IsTruncated == true); + + return files; + } + + /// True for a single-segment file name ending in .yaml or .yml. + public static bool IsYamlFileName(string file) => + file.Length > 0 + && (file.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase) || file.EndsWith(".yml", StringComparison.OrdinalIgnoreCase)); +} diff --git a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs index a70af45507..b67fd3a0a7 100644 --- a/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs +++ b/src/services/Elastic.Changelog/Scrubbing/ScrubberProcessor.cs @@ -29,7 +29,8 @@ public sealed class ScrubberProcessor( string publicBucketName, IChangelogContentScrubber scrubber, RegistryReconciler reconciler, - ReconcileMetrics? metrics = null + ReconcileMetrics? metrics = null, + string? privateBucketName = null ) { // Bounds the reread-and-redo loop of post-write source validation. Each redo only triggers @@ -50,6 +51,9 @@ private sealed class GroupWork(ChangelogScope scope) { public ChangelogScope Scope { get; } = scope; public HashSet MessageIds { get; } = [with(StringComparer.Ordinal)]; + + /// Set by an explicit reconcile message: object-reconcile the union of both buckets' listings first. + public bool FullHeal { get; set; } } /// @@ -66,6 +70,25 @@ public async Task> ProcessAsync(IReadOnlyList> ProcessAsync(IReadOnlyList groupWork, ChangelogScope scope, string messageId) + private static GroupWork AddGroup(Dictionary groupWork, ChangelogScope scope, string messageId) { if (!groupWork.TryGetValue(scope.Prefix, out var work)) { @@ -168,6 +193,36 @@ private static void AddGroup(Dictionary groupWork, ChangelogS groupWork[scope.Prefix] = work; } _ = work.MessageIds.Add(messageId); + return work; + } + + /// + /// The full group heal an explicit reconcile message asks for: object-level reconcile over the + /// union of both buckets' group listings — scrub/copy what is live in the private bucket, + /// delete what is not — so lost or DLQ-expired scrub events are recoverable by message. Keys + /// already reconciled by this batch's own S3 events are skipped. + /// + private async Task HealGroupObjectsAsync(ChangelogScope scope, IReadOnlyCollection alreadyReconciled, Cancel ctx) + { + if (privateBucketName is null) + { + throw new InvalidOperationException( + "A reconcile message arrived but no private bucket is configured (PRIVATE_BUCKET_NAME); cannot perform a full group heal."); + } + + var privateObjects = await S3GroupListing.ListImmediateYamlObjectsAsync(s3Client, privateBucketName, scope, ctx); + var publicObjects = await S3GroupListing.ListImmediateYamlObjectsAsync(s3Client, publicBucketName, scope, ctx); + var keys = new SortedSet(StringComparer.Ordinal); + keys.UnionWith(privateObjects.Select(o => o.Key)); + keys.UnionWith(publicObjects.Select(o => o.Key)); + + _logger.LogInformation("Full heal of {Scope}: {Count} key(s) in the union of both buckets", scope, keys.Count); + + foreach (var key in keys.Where(k => !alreadyReconciled.Contains(k))) + { + ctx.ThrowIfCancellationRequested(); + await ReconcileObjectAsync(privateBucketName, key, ctx); + } } /// diff --git a/src/services/Elastic.Changelog/Uploading/Registry.cs b/src/services/Elastic.Changelog/Uploading/Registry.cs index ff81cb0b4c..48e5ff9bbd 100644 --- a/src/services/Elastic.Changelog/Uploading/Registry.cs +++ b/src/services/Elastic.Changelog/Uploading/Registry.cs @@ -83,6 +83,9 @@ public sealed record RegistryBundle /// 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. /// + // The snake_case policy would emit "e_tag"; the documented manifest format — and what the + // consumer-side ChangelogRegistryBundle reads — is "etag". + [JsonPropertyName("etag")] public required string ETag { get; init; } } diff --git a/src/tooling/docs-builder/Commands/ChangelogRegistryCommands.cs b/src/tooling/docs-builder/Commands/ChangelogRegistryCommands.cs new file mode 100644 index 0000000000..d36ee438af --- /dev/null +++ b/src/tooling/docs-builder/Commands/ChangelogRegistryCommands.cs @@ -0,0 +1,115 @@ +// 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.Reconciliation; +using Elastic.Documentation.Diagnostics; +using Elastic.Documentation.Services; +using Microsoft.Extensions.Logging; +using Nullean.Argh; +using Nullean.Argh.Documentation; + +namespace Documentation.Builder.Commands; + +/// Operate on the scrubber-owned public changelog registries. +internal sealed class ChangelogRegistryCommands( + ILoggerFactory logFactory, + IDiagnosticsCollector collector +) +{ + /// + /// Send explicit reconcile messages to the scrubber queue so the Lambda performs a full group + /// heal (object-level reconcile over the union of both buckets, then a registry rebuild from + /// public state) for every planned group. This command never mutates S3 itself — the scrubber + /// Lambda stays the public bucket's single writer. Convergent: re-running re-plans against + /// current state. Enqueuing is not reconciling — gate on `changelog registry verify` after the + /// queue drains and the DLQ is empty. + /// + /// Private changelog bundles bucket to plan from. + /// Public (CDN) changelog bundles bucket to plan from. + /// URL of the scrubber SQS queue to send reconcile messages to. + /// Only reconcile this bundle group (bundle/{product}/). Mutually exclusive with --owner/--repo/--branch. + /// GitHub owner of a single changelog-pool group to reconcile. + /// Repository of a single changelog-pool group to reconcile. + /// Branch of a single changelog-pool group to reconcile (verbatim; slashes allowed). + /// Print the group plan without sending anything. + /// Skip the interactive confirmation (required when stdin is not a terminal). + [RequiresAuth] + [CommandIntent(Intent.Destructive | Intent.Idempotent)] + [MutationScope(MutationScope.Global)] + [NoOptionsInjection] + public async Task Reconcile( + string s3BucketName, + string publicS3BucketName, + string queueUrl, + string? product = null, + string? owner = null, + string? repo = null, + string? branch = null, + [DryRun] bool dryRun = false, + bool yes = false, + Cancel ct = default + ) + { + await using var serviceInvoker = new ServiceInvoker(collector); + var service = new ChangelogRegistryReconcileService(logFactory); + var args = new ChangelogRegistryReconcileArguments + { + S3BucketName = s3BucketName, + PublicS3BucketName = publicS3BucketName, + QueueUrl = queueUrl, + Product = product, + Owner = owner, + Repo = repo, + Branch = branch, + DryRun = dryRun, + AssumeYes = yes + }; + serviceInvoker.AddCommand(service, args, + static async (s, c, state, ct) => await s.Reconcile(c, state, ct) + ); + return await serviceInvoker.InvokeAsync(ct); + } + + /// + /// Compare each planned group's public registry.json against what a reconcile of the + /// current public listing would write, and report divergence (missing, stale, corrupt, + /// object-divergent; unsupported schema reported distinctly). Strictly read-only. Zero + /// divergence across the plan is the cutover completion gate — and the standing + /// drift-diagnosis tool afterwards. + /// + /// Private changelog bundles bucket to plan from. + /// Public (CDN) changelog bundles bucket to verify. + /// Only verify this bundle group (bundle/{product}/). Mutually exclusive with --owner/--repo/--branch. + /// GitHub owner of a single changelog-pool group to verify. + /// Repository of a single changelog-pool group to verify. + /// Branch of a single changelog-pool group to verify (verbatim; slashes allowed). + [RequiresAuth] + [NoOptionsInjection] + public async Task Verify( + string s3BucketName, + string publicS3BucketName, + string? product = null, + string? owner = null, + string? repo = null, + string? branch = null, + Cancel ct = default + ) + { + await using var serviceInvoker = new ServiceInvoker(collector); + var service = new ChangelogRegistryVerifyService(logFactory); + var args = new ChangelogRegistryVerifyArguments + { + S3BucketName = s3BucketName, + PublicS3BucketName = publicS3BucketName, + Product = product, + Owner = owner, + Repo = repo, + Branch = branch + }; + serviceInvoker.AddCommand(service, args, + static async (s, c, state, ct) => await s.Verify(c, state, ct) + ); + return await serviceInvoker.InvokeAsync(ct); + } +} diff --git a/src/tooling/docs-builder/Program.cs b/src/tooling/docs-builder/Program.cs index 98eb0bc178..2d1001a92d 100644 --- a/src/tooling/docs-builder/Program.cs +++ b/src/tooling/docs-builder/Program.cs @@ -49,7 +49,10 @@ _ = app.Map(); _ = app.Map(); _ = app.Map(); - _ = app.MapNamespace("changelog"); + _ = app.MapNamespace("changelog", g => + { + _ = g.MapNamespace("registry"); + }); _ = app.MapNamespace("inbound-links"); _ = app.Map(); diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/ChangelogRegistryServiceTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/ChangelogRegistryServiceTests.cs new file mode 100644 index 0000000000..bac857180e --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Reconciliation/ChangelogRegistryServiceTests.cs @@ -0,0 +1,180 @@ +// 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 Amazon.SQS; +using Amazon.SQS.Model; +using AwesomeAssertions; +using Elastic.Changelog.Reconciliation; +using FakeItEasy; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Elastic.Changelog.Tests.Reconciliation; + +[SuppressMessage("Usage", "CA1001:Types that own disposable fields should be disposable")] +public class ChangelogRegistryServiceTests +{ + private const string PrivateBucket = "private-bucket"; + private const string PublicBucket = "public-bucket"; + + private readonly FakeS3 _s3 = new(PrivateBucket, PublicBucket); + private readonly IAmazonSQS _sqs = A.Fake(); + private readonly List _sent = []; + private readonly TestDiagnosticsCollector _collector; + + public ChangelogRegistryServiceTests(ITestOutputHelper output) + { + _collector = new TestDiagnosticsCollector(output); + _ = A.CallTo(() => _sqs.SendMessageAsync(A._, A._)) + .Invokes((SendMessageRequest r, CancellationToken _) => _sent.Add(r)) + .ReturnsLazily((SendMessageRequest _, CancellationToken _) => + Task.FromResult(new SendMessageResponse { MessageId = $"sqs-{_sent.Count}" })); + } + + private ChangelogRegistryReconcileService ReconcileService(Func? confirm = null) => + new(NullLoggerFactory.Instance, _s3.Client, _sqs, confirm ?? (() => "y")); + + private static ChangelogRegistryReconcileArguments ReconcileArgs( + bool dryRun = false, bool yes = false, string? product = null) => new() + { + S3BucketName = PrivateBucket, + PublicS3BucketName = PublicBucket, + QueueUrl = "https://sqs.example/queue", + DryRun = dryRun, + AssumeYes = yes, + Product = product + }; + + private Cancel Ctx => TestContext.Current.CancellationToken; + + [Fact] + public async Task Reconcile_DiscoversTheUnionOfBothBuckets_IncludingOrphanPublicGroups() + { + // Private: one bundle group and one pool. Public: an orphan group (nothing private backs + // it) that only a union discovery would plan — its reconcile is what deletes the leftovers. + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "a"); + _ = _s3.Seed(PrivateBucket, "changelog/elastic/repo/main/entry.yaml", "b"); + _ = _s3.Seed(PublicBucket, "bundle/orphaned/old.yaml", "c"); + _ = _s3.Seed(PublicBucket, "bundle/manifest-only/registry.json", "{}"); + + var ok = await ReconcileService().Reconcile(_collector, ReconcileArgs(yes: true), Ctx); + + ok.Should().BeTrue(); + var bodies = _sent.Select(r => r.MessageBody).ToList(); + bodies.Should().HaveCount(4); + bodies.Should().Contain(b => b.Contains("\"scope\":\"bundle\"", StringComparison.Ordinal) && b.Contains("\"group\":\"elasticsearch\"", StringComparison.Ordinal)); + bodies.Should().Contain(b => b.Contains("\"scope\":\"changelog\"", StringComparison.Ordinal) && b.Contains("\"group\":\"elastic/repo/main\"", StringComparison.Ordinal)); + bodies.Should().Contain(b => b.Contains("\"group\":\"orphaned\"", StringComparison.Ordinal)); + bodies.Should().Contain(b => b.Contains("\"group\":\"manifest-only\"", StringComparison.Ordinal), + "a group holding only an orphaned manifest still needs its reconcile"); + bodies.Should().OnlyContain(b => b.Contains("\"kind\":\"reconcile\"", StringComparison.Ordinal) && b.Contains("\"version\":1", StringComparison.Ordinal)); + } + + [Fact] + public async Task Reconcile_EveryMessageValidatesAndSharesOneCorrelationId() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "a"); + _ = _s3.Seed(PrivateBucket, "bundle/kibana/kb-9.1.0.yaml", "b"); + + var ok = await ReconcileService().Reconcile(_collector, ReconcileArgs(yes: true), Ctx); + + ok.Should().BeTrue(); + var correlationIds = new HashSet(StringComparer.Ordinal); + foreach (var request in _sent) + { + _ = ReconcileQueueMessage.TryRead(request.MessageBody, out var message); + message.Should().NotBeNull(); + message!.TryResolveScope(out _, out _).Should().BeTrue("the CLI must only ever send messages the Lambda accepts"); + _ = correlationIds.Add(message.CorrelationId!); + } + correlationIds.Should().ContainSingle("one run stamps one correlation id on its whole ledger"); + } + + [Fact] + public async Task Reconcile_DryRun_SendsNothing() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "a"); + + var ok = await ReconcileService().Reconcile(_collector, ReconcileArgs(dryRun: true), Ctx); + + ok.Should().BeTrue(); + _sent.Should().BeEmpty(); + } + + [Fact] + public async Task Reconcile_DeclinedConfirmation_Aborts() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "a"); + + var ok = await ReconcileService(confirm: () => "n").Reconcile(_collector, ReconcileArgs(), Ctx); + + ok.Should().BeFalse(); + _sent.Should().BeEmpty(); + } + + [Fact] + public async Task Reconcile_SingleScopeFilter_PlansOnlyThatGroup() + { + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "a"); + _ = _s3.Seed(PrivateBucket, "bundle/kibana/kb-9.1.0.yaml", "b"); + + var ok = await ReconcileService().Reconcile(_collector, ReconcileArgs(yes: true, product: "kibana"), Ctx); + + ok.Should().BeTrue(); + _sent.Should().ContainSingle().Which.MessageBody.Should().Contain("\"group\":\"kibana\""); + } + + [Fact] + public async Task Reconcile_MixedScopeForms_IsRejected() + { + var args = ReconcileArgs(yes: true, product: "kibana") with { Owner = "elastic", Repo = "repo", Branch = "main" }; + + var ok = await ReconcileService().Reconcile(_collector, args, Ctx); + + ok.Should().BeFalse(); + _sent.Should().BeEmpty(); + } + + [Fact] + public async Task Verify_ReportsDivergenceAcrossThePlanAndFails() + { + // elasticsearch is converged; kibana has objects but no manifest. + var yaml = "products:\n - product: elasticsearch\n target: 9.1.0\n repo: elasticsearch\n owner: elastic\n"; + var etag = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", yaml); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/registry.json", + "{\"schema_version\":1,\"product\":\"elasticsearch\",\"producer\":\"" + RegistryReconciler.Producer + + "\",\"generated_at\":\"2026-07-01T00:00:00+00:00\",\"bundles\":[{\"file\":\"es-9.1.0.yaml\",\"target\":\"9.1.0\",\"etag\":\"" + etag + "\"}]}"); + _ = _s3.Seed(PublicBucket, "bundle/kibana/kb-9.1.0.yaml", "kb"); + + var service = new ChangelogRegistryVerifyService(NullLoggerFactory.Instance, _s3.Client); + var ok = await service.Verify(_collector, new ChangelogRegistryVerifyArguments + { + S3BucketName = PrivateBucket, + PublicS3BucketName = PublicBucket + }, Ctx); + + ok.Should().BeFalse("kibana diverges"); + _s3.Puts.Should().BeEmpty("verify is strictly read-only"); + _s3.Deletes.Should().BeEmpty(); + } + + [Fact] + public async Task Verify_ConvergedPlan_Succeeds() + { + var yaml = "products:\n - product: elasticsearch\n target: 9.1.0\n repo: elasticsearch\n owner: elastic\n"; + var etag = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", yaml); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/registry.json", + "{\"schema_version\":1,\"product\":\"elasticsearch\",\"producer\":\"" + RegistryReconciler.Producer + + "\",\"generated_at\":\"2026-07-01T00:00:00+00:00\",\"bundles\":[{\"file\":\"es-9.1.0.yaml\",\"target\":\"9.1.0\",\"etag\":\"" + etag + "\"}]}"); + + var service = new ChangelogRegistryVerifyService(NullLoggerFactory.Instance, _s3.Client); + var ok = await service.Verify(_collector, new ChangelogRegistryVerifyArguments + { + S3BucketName = PrivateBucket, + PublicS3BucketName = PublicBucket + }, Ctx); + + ok.Should().BeTrue(); + } +} diff --git a/tests/Elastic.Changelog.Tests/Reconciliation/RegistryVerifyTests.cs b/tests/Elastic.Changelog.Tests/Reconciliation/RegistryVerifyTests.cs new file mode 100644 index 0000000000..b15b9f027c --- /dev/null +++ b/tests/Elastic.Changelog.Tests/Reconciliation/RegistryVerifyTests.cs @@ -0,0 +1,185 @@ +// 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 RegistryVerifyTests +{ + private const string PublicBucket = "public-bucket"; + + private readonly FakeS3 _s3 = new(PublicBucket); + private readonly RegistryReconciler _reconciler; + + public RegistryVerifyTests() => + _reconciler = new RegistryReconciler(NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero); + + private static ChangelogScope Scope + { + get + { + _ = ChangelogScope.TryCreateBundle("elasticsearch", out var scope); + return scope!; + } + } + + // language=yaml + private const 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 + """; + + private void SeedManifest(int schemaVersion = Registry.CurrentSchemaVersion, string? producer = RegistryReconciler.Producer, params RegistryBundle[] bundles) + { + var manifest = new Registry + { + SchemaVersion = schemaVersion, + Product = "elasticsearch", + Producer = producer, + GeneratedAt = new DateTimeOffset(2026, 7, 1, 0, 0, 0, TimeSpan.Zero), + Bundles = bundles + }; + _ = _s3.Seed(PublicBucket, Scope.RegistryKey, JsonSerializer.Serialize(manifest, RegistryJsonContext.Default.Registry)); + } + + private Cancel Ctx => TestContext.Current.CancellationToken; + + [Fact] + public async Task Verify_ConvergedGroup_ReportsNothingAndWritesNothing() + { + var etag = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + SeedManifest(bundles: new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + divergences.Should().BeEmpty(); + _s3.Puts.Should().BeEmpty(); + _s3.Deletes.Should().BeEmpty(); + } + + [Fact] + public async Task Verify_ObjectsWithoutManifest_ReportsMissingManifest() + { + _ = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + var finding = divergences.Should().ContainSingle().Subject; + finding.Kind.Should().Be(RegistryDivergenceKind.Missing); + finding.File.Should().Be("registry.json"); + } + + [Fact] + public async Task Verify_ManifestForEmptyGroup_ReportsStaleManifest() + { + SeedManifest(); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + var finding = divergences.Should().ContainSingle().Subject; + finding.Kind.Should().Be(RegistryDivergenceKind.Stale); + finding.File.Should().Be("registry.json"); + } + + [Fact] + public async Task Verify_UnlistedObject_ReportsMissingEntry() + { + var etag = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + _ = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.2.0.yaml", BundleYaml); + SeedManifest(bundles: new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + var finding = divergences.Should().ContainSingle().Subject; + finding.Kind.Should().Be(RegistryDivergenceKind.Missing); + finding.File.Should().Be("es-9.2.0.yaml"); + } + + [Fact] + public async Task Verify_EntryWhoseObjectIsGone_ReportsStaleEntry() + { + var etag = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + SeedManifest(bundles: + [ + new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }, + new RegistryBundle { File = "gone.yaml", Target = "9.0.0", ETag = "dead" } + ]); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + var finding = divergences.Should().ContainSingle().Subject; + finding.Kind.Should().Be(RegistryDivergenceKind.Stale); + finding.File.Should().Be("gone.yaml"); + } + + [Fact] + public async Task Verify_EntryWithWrongETag_ReportsObjectDivergent() + { + _ = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + SeedManifest(bundles: new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = "outdated" }); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + var finding = divergences.Should().ContainSingle().Subject; + finding.Kind.Should().Be(RegistryDivergenceKind.ObjectDivergent); + finding.File.Should().Be("es-9.1.0.yaml"); + } + + [Fact] + public async Task Verify_CorruptManifest_ReportsCorrupt() + { + _ = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + _ = _s3.Seed(PublicBucket, Scope.RegistryKey, "{ not json "); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + divergences.Should().ContainSingle().Subject.Kind.Should().Be(RegistryDivergenceKind.Corrupt); + } + + [Fact] + public async Task Verify_NewerSchema_ReportsUnsupportedSchemaDistinctly() + { + _ = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + SeedManifest(schemaVersion: Registry.CurrentSchemaVersion + 1); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + divergences.Should().ContainSingle().Subject.Kind.Should().Be(RegistryDivergenceKind.UnsupportedSchema); + } + + [Fact] + public async Task Verify_LegacyProducerManifest_ReportsStaleMetadataEvenWhenEntriesMatch() + { + var etag = _s3.Seed(PublicBucket, Scope.Prefix + "es-9.1.0.yaml", BundleYaml); + SeedManifest(producer: null, bundles: new RegistryBundle { File = "es-9.1.0.yaml", Target = "9.1.0", ETag = etag }); + + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + var finding = divergences.Should().ContainSingle().Subject; + finding.Kind.Should().Be(RegistryDivergenceKind.Stale); + finding.File.Should().Be("registry.json"); + } + + [Fact] + public async Task Verify_EmptyGroupWithoutManifest_ReportsNothing() + { + var divergences = await _reconciler.VerifyGroupAsync(Scope, Ctx); + + divergences.Should().BeEmpty(); + } +} diff --git a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs index 5fb95d40b6..0b8a1a3923 100644 --- a/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs +++ b/tests/Elastic.Changelog.Tests/Scrubbing/ScrubberProcessorTests.cs @@ -33,7 +33,7 @@ public ScrubberProcessorTests() var reconciler = new RegistryReconciler( NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); _processor = new ScrubberProcessor( - NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, _metrics); + NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, _metrics, PrivateBucket); } private Cancel Ctx => TestContext.Current.CancellationToken; @@ -269,6 +269,101 @@ public async Task Process_BatchMixingObjectAndRegistryEvents_MarksGroupContribut _metrics.GroupReconciles.Should().Be(1); } + private static ScrubberQueueMessage Reconcile(string scope, string group, int version = 1) + { + var id = $"msg-{Interlocked.Increment(ref MessageCounter)}"; + var body = "{\"kind\":\"reconcile\",\"version\":" + version + ",\"scope\":\"" + scope + + "\",\"group\":\"" + group + "\",\"correlation_id\":\"test-run\"}"; + return new ScrubberQueueMessage(id, body); + } + + [Fact] + public async Task Process_ReconcileMessage_PerformsAFullGroupHeal() + { + // A full heal reconciles the union of both buckets: live private objects are (re)copied, + // the orphan public object is deleted, and the manifest is rebuilt — recovering drift no + // pending S3 event would ever repair (lost/DLQ-expired scrub events). + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.1.0.yaml", "one"); + _ = _s3.Seed(PrivateBucket, "bundle/elasticsearch/es-9.2.0.yaml", "two"); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml", "stale-copy"); + _ = _s3.Seed(PublicBucket, "bundle/elasticsearch/orphan.yaml", "orphan"); + + var failed = await _processor.ProcessAsync([Reconcile("bundle", "elasticsearch")], Ctx); + + failed.Should().BeEmpty(); + _s3.ContentOf(PublicBucket, "bundle/elasticsearch/es-9.1.0.yaml").Should().Be("scrubbed: one"); + _s3.ContentOf(PublicBucket, "bundle/elasticsearch/es-9.2.0.yaml").Should().Be("scrubbed: two"); + _s3.Exists(PublicBucket, "bundle/elasticsearch/orphan.yaml").Should().BeFalse("nothing in the private bucket backs it"); + PublicManifest("bundle/elasticsearch/registry.json").Bundles.Select(b => b.File) + .Should().BeEquivalentTo(["es-9.1.0.yaml", "es-9.2.0.yaml"]); + } + + [Fact] + public async Task Process_ReconcileMessage_ForAGroupEmptyInBothBuckets_DeletesTheManifest() + { + _ = _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([Reconcile("bundle", "elasticsearch")], Ctx); + + failed.Should().BeEmpty(); + _s3.Exists(PublicBucket, "bundle/elasticsearch/registry.json").Should().BeFalse(); + } + + [Theory] + [InlineData("bundle", "../escape")] + [InlineData("bundle", "")] + [InlineData("changelog", "elastic/repo")] + [InlineData("changelog", "elastic/re po/main")] + [InlineData("pool", "elasticsearch")] + public async Task Process_MalformedReconcileMessage_IsRejected(string scope, string group) + { + var message = Reconcile(scope, group); + + var failed = await _processor.ProcessAsync([message], Ctx); + + failed.Should().ContainSingle().Which.Should().Be(message.MessageId); + _s3.Puts.Should().BeEmpty(); + _s3.Deletes.Should().BeEmpty(); + } + + [Fact] + public async Task Process_ReconcileMessageWithUnsupportedVersion_IsRejected() + { + var message = Reconcile("bundle", "elasticsearch", version: 2); + + var failed = await _processor.ProcessAsync([message], Ctx); + + failed.Should().ContainSingle().Which.Should().Be(message.MessageId); + } + + [Fact] + public async Task Process_ReconcileMessage_WithoutPrivateBucketConfigured_FailsTheMessage() + { + var reconciler = new RegistryReconciler( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, retryBaseDelay: TimeSpan.Zero, metrics: _metrics); + var processor = new ScrubberProcessor( + NullLoggerFactory.Instance, _s3.Client, PublicBucket, _scrubber, reconciler, _metrics); + var message = Reconcile("bundle", "elasticsearch"); + + var failed = await processor.ProcessAsync([message], Ctx); + + failed.Should().ContainSingle().Which.Should().Be(message.MessageId); + } + + [Fact] + public async Task Process_ReconcileMessage_ForABranchWithSlashes_HealsTheRightPool() + { + _ = _s3.Seed(PrivateBucket, "changelog/elastic/repo/main/feature/entry.yaml", "entry"); + + var failed = await _processor.ProcessAsync([Reconcile("changelog", "elastic/repo/main/feature")], Ctx); + + failed.Should().BeEmpty(); + _s3.Exists(PublicBucket, "changelog/elastic/repo/main/feature/entry.yaml").Should().BeTrue(); + PublicManifest("changelog/elastic/repo/main/feature/registry.json").Bundles.Select(b => b.File) + .Should().Equal("entry.yaml"); + } + // language=yaml private static string BundleYaml() => """ products: