diff --git a/src/MUI.Crawler/Persistence/NpgsqlLeadLog.cs b/src/MUI.Crawler/Persistence/NpgsqlLeadLog.cs index 1431523..307ecc4 100644 --- a/src/MUI.Crawler/Persistence/NpgsqlLeadLog.cs +++ b/src/MUI.Crawler/Persistence/NpgsqlLeadLog.cs @@ -102,6 +102,16 @@ UPDATE crawl_lead cancellationToken: ct)); } + public async Task AbandonAsync(Guid id, CancellationToken ct) + { + await using var connection = await dataSource.OpenConnectionAsync(ct); + + await connection.ExecuteAsync(new CommandDefinition( + "DELETE FROM crawl_lead WHERE id = @id AND outcome = 'pending' AND crawl_target_id IS NULL", + new { id }, + cancellationToken: ct)); + } + public async Task> RecentAsync(int limit, CancellationToken ct) { await using var connection = await dataSource.OpenConnectionAsync(ct); diff --git a/src/MUI.Discovery/Intake/Lead.cs b/src/MUI.Discovery/Intake/Lead.cs index 0f003f0..d8453ee 100644 --- a/src/MUI.Discovery/Intake/Lead.cs +++ b/src/MUI.Discovery/Intake/Lead.cs @@ -67,6 +67,14 @@ Task CompleteAsync( Guid? crawlTargetId, CancellationToken ct); + /// + /// Removes a reservation that never got further than being taken, so it stops holding a slot. + /// + /// + /// Only a row still pending: a lead that reached an outcome is a record and stays one. + /// + Task AbandonAsync(Guid id, CancellationToken ct); + /// The newest leads first, so the routine can skip pages it has already read. Task> RecentAsync(int limit, CancellationToken ct); } @@ -157,40 +165,58 @@ public async Task SubmitAsync( return new LeadReceipt(SubmissionOutcome.TooMany); } - if (!SubmittedAddressReader.TryRead(host, port.ToString(CultureInfo.InvariantCulture), out var address)) + // Set the moment a target exists, which is the line between a reservation that may be given + // back and one that may not. + Guid? planted = null; + + try { - return await CompleteAsync(dryRun, reservation, new LeadReceipt(SubmissionOutcome.Malformed), null, ct); - } + if (!SubmittedAddressReader.TryRead(host, port.ToString(CultureInfo.InvariantCulture), out var address)) + { + return await CompleteAsync(dryRun, reservation, new LeadReceipt(SubmissionOutcome.Malformed), null, ct); + } - var ruling = await intake.RuleAsync(address, ct); + var ruling = await intake.RuleAsync(address, ct); + + if (ruling.Refusal is { } refusal) + { + return await CompleteAsync( + dryRun, reservation, new LeadReceipt(refusal, address, ruling.GameId, ruling.Detail), null, ct); + } + + if (dryRun) + { + return new LeadReceipt(SubmissionOutcome.Accepted, address); + } + + // Due now: an announcement is a game that has just opened, and the probe is what decides + // whether it is one. IsOperatorSeed stays false — §7.2's exemption is never inferred. + planted = await targets.AddAsync( + new CrawlTarget + { + Id = Guid.CreateVersion7(), + Host = address.Host, + Port = address.Port, + NextProbeAt = now, + FirstSeenAt = now, + SubmittedAt = now, + DiscoveredVia = DiscoverySource.Announcement, + }, + ct); - if (ruling.Refusal is { } refusal) - { return await CompleteAsync( - dryRun, reservation, new LeadReceipt(refusal, address, ruling.GameId, ruling.Detail), null, ct); + dryRun, reservation, new LeadReceipt(SubmissionOutcome.Accepted, address), planted, ct); } - - if (dryRun) + catch when (!dryRun && planted is null) { - return new LeadReceipt(SubmissionOutcome.Accepted, address); + // Nothing was written but the reservation, so give it back: left pending, it would hold a + // slot until the window passed and tell the routine, through crawl_leads, that a page it + // never finished had been handed in. Once a target exists the row stays, pending, as the + // only record of the page that produced it — handing the page in again then answers + // already_queued and completes normally. + await AbandonQuietlyAsync(reservation); + throw; } - - // Due now: an announcement is a game that has just opened, and the probe is what decides - // whether it is one. IsOperatorSeed stays false — §7.2's exemption is never inferred. - var id = await targets.AddAsync( - new CrawlTarget - { - Id = Guid.CreateVersion7(), - Host = address.Host, - Port = address.Port, - NextProbeAt = now, - FirstSeenAt = now, - SubmittedAt = now, - DiscoveredVia = DiscoverySource.Announcement, - }, - ct); - - return await CompleteAsync(dryRun, reservation, new LeadReceipt(SubmissionOutcome.Accepted, address), id, ct); } /// @@ -232,6 +258,26 @@ private static void RequireWebPage(Uri url, string name) } } + /// + /// Gives a reservation back, without letting a second failure hide the first. + /// + /// + /// Not cancellable: this runs because the call already failed, quite possibly by cancellation, and + /// a token that has fired would leave exactly the row this exists to remove. + /// + private async Task AbandonQuietlyAsync(Guid reservation) + { + try + { + await log.AbandonAsync(reservation, CancellationToken.None); + } + catch (Exception) + { + // The original failure is the one worth surfacing. A row this could not remove reads as + // pending in crawl_leads, which tells the routine to hand the page in again. + } + } + private async Task CompleteAsync( bool dryRun, Guid reservation, LeadReceipt receipt, Guid? targetId, CancellationToken ct) { diff --git a/src/MUI.Web/Mcp/CrawlAdminTools.cs b/src/MUI.Web/Mcp/CrawlAdminTools.cs index 792a0f6..194a1be 100644 --- a/src/MUI.Web/Mcp/CrawlAdminTools.cs +++ b/src/MUI.Web/Mcp/CrawlAdminTools.cs @@ -153,7 +153,9 @@ public async Task CrawlLeadAddAsync( [Description(""" The leads handed in through crawl_lead_add, newest first: the address, the page it was read from, and what became of it. The lead routine reads this first so it does not hand in a page - it has already read. Our own note about our own crawl -- none of it reaches a game page. + it has already read -- except a row whose outcome is `pending`, which never finished and should + be handed in again (a second try answers already_queued if the first got as far as the + registry). Our own note about our own crawl -- none of it reaches a game page. """)] public async Task> CrawlLeadsAsync( [Description("How many leads to list. Default 100.")] int batch = 100, diff --git a/tests/MUI.Crawler.Tests/SubmissionPostgresTests.Leads.cs b/tests/MUI.Crawler.Tests/SubmissionPostgresTests.Leads.cs index 41f0cc9..1ba9eb3 100644 --- a/tests/MUI.Crawler.Tests/SubmissionPostgresTests.Leads.cs +++ b/tests/MUI.Crawler.Tests/SubmissionPostgresTests.Leads.cs @@ -183,6 +183,31 @@ await Assert.That(await log.TryBeginAsync( .IsFalse(); } + /// + /// Abandoning gives back a bare reservation and never touches a lead that reached an outcome. + /// + [Test] + public async Task AbandoningRemovesOnlyAReservationThatWentNowhere() + { + await using var database = await PostgresFixture.MigratedAsync(); + var log = new NpgsqlLeadLog(database.DataSource); + var now = DateTimeOffset.UtcNow; + var bare = Guid.CreateVersion7(); + var finished = Guid.CreateVersion7(); + + await log.TryBeginAsync(bare, Announcement, now, 10, now.AddDays(-1), None); + await log.TryBeginAsync(finished, Announcement, now, 10, now.AddDays(-1), None); + await log.CompleteAsync( + finished, new SubmittedAddress("mud.example.org", 4201), SubmissionOutcome.AlreadyQueued, null, None); + + await log.AbandonAsync(bare, None); + await log.AbandonAsync(finished, None); + + var left = await log.RecentAsync(10, None); + + await Assert.That(left.Select(r => r.Id)).IsEquivalentTo([finished]); + } + /// The bound holds under a concurrent burst, as the form's does. [Test] public async Task AConcurrentBurstOfLeadsDoesNotWalkThroughTheBound() diff --git a/tests/MUI.Discovery.Tests/Intake/LeadServiceTests.cs b/tests/MUI.Discovery.Tests/Intake/LeadServiceTests.cs index 258062b..a447fac 100644 --- a/tests/MUI.Discovery.Tests/Intake/LeadServiceTests.cs +++ b/tests/MUI.Discovery.Tests/Intake/LeadServiceTests.cs @@ -216,6 +216,51 @@ public async Task TheWindowTakesOnlySoManyLeadsAndThenReopens() await Assert.That(tomorrow.Outcome).IsEqualTo(SubmissionOutcome.Accepted); } + /// + /// A failure before anything was written gives the reservation back: it holds no slot and does not + /// tell the routine, through crawl_leads, that a page it never finished was handed in. + /// + [Test] + public async Task AFailureBeforeATargetExistsReleasesTheReservation() + { + var world = Build(new LeadOptions { PerWindow = 1 }); + world.Log.FailNextComplete = true; + + // Unreadable, so the only write after the reservation is the one that fails. + await Assert.That(async () => await world.Service.SubmitAsync("", 4201, Post, ct: None)) + .ThrowsException(); + + await Assert.That(world.Log.Rows).IsEmpty(); + + // The one slot is still there to take. + var next = await world.Service.SubmitAsync("mud.example.org", 4201, Post, ct: None); + + await Assert.That(next.Outcome).IsEqualTo(SubmissionOutcome.Accepted); + } + + /// + /// Once a target exists the row is never given back, even if completing it fails: it is the only + /// record of the page that produced that target, and handing the page in again settles it. + /// + [Test] + public async Task AFailureAfterTheTargetExistsKeepsThePendingRow() + { + var world = Build(); + world.Log.FailNextComplete = true; + + await Assert.That(async () => await world.Service.SubmitAsync("mud.example.org", 4201, Post, ct: None)) + .ThrowsException(); + + var row = world.Log.Rows.Single(); + + await Assert.That(row.Outcome).IsNull(); + await Assert.That(await world.Targets.ByAddressAsync("mud.example.org", 4201, None)).IsNotNull(); + + var again = await world.Service.SubmitAsync("mud.example.org", 4201, Post, ct: None); + + await Assert.That(again.Outcome).IsEqualTo(SubmissionOutcome.AlreadyQueued); + } + /// An address that cannot be read is recorded as that, and fabricates no default port. [Test] [Arguments("", 4201)] diff --git a/tests/MUI.Discovery.Tests/Support/InMemoryLeadLog.cs b/tests/MUI.Discovery.Tests/Support/InMemoryLeadLog.cs index 6dd8f7a..1cea751 100644 --- a/tests/MUI.Discovery.Tests/Support/InMemoryLeadLog.cs +++ b/tests/MUI.Discovery.Tests/Support/InMemoryLeadLog.cs @@ -39,6 +39,9 @@ public Task TryBeginAsync( } } + /// Makes the next throw, as a dropped connection would. + public bool FailNextComplete { get; set; } + public Task CompleteAsync( Guid id, SubmittedAddress? address, @@ -48,6 +51,12 @@ public Task CompleteAsync( { lock (_gate) { + if (FailNextComplete) + { + FailNextComplete = false; + throw new InvalidOperationException("connection lost"); + } + var index = _rows.FindIndex(r => r.Id == id); if (index >= 0) @@ -65,6 +74,16 @@ public Task CompleteAsync( return Task.CompletedTask; } + public Task AbandonAsync(Guid id, CancellationToken ct) + { + lock (_gate) + { + _rows.RemoveAll(r => r.Id == id && r.Outcome is null && r.CrawlTargetId is null); + } + + return Task.CompletedTask; + } + public Task> RecentAsync(int limit, CancellationToken ct) { lock (_gate)