Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/MUI.Crawler/Persistence/NpgsqlLeadLog.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<IReadOnlyList<LeadRecord>> RecentAsync(int limit, CancellationToken ct)
{
await using var connection = await dataSource.OpenConnectionAsync(ct);
Expand Down
100 changes: 73 additions & 27 deletions src/MUI.Discovery/Intake/Lead.cs
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,14 @@ Task CompleteAsync(
Guid? crawlTargetId,
CancellationToken ct);

/// <summary>
/// Removes a reservation that never got further than being taken, so it stops holding a slot.
/// </summary>
/// <remarks>
/// Only a row still <c>pending</c>: a lead that reached an outcome is a record and stays one.
/// </remarks>
Task AbandonAsync(Guid id, CancellationToken ct);

/// <summary>The newest leads first, so the routine can skip pages it has already read.</summary>
Task<IReadOnlyList<LeadRecord>> RecentAsync(int limit, CancellationToken ct);
}
Expand Down Expand Up @@ -157,40 +165,58 @@ public async Task<LeadReceipt> 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);
}

/// <summary>
Expand Down Expand Up @@ -232,6 +258,26 @@ private static void RequireWebPage(Uri url, string name)
}
}

/// <summary>
/// Gives a reservation back, without letting a second failure hide the first.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
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<LeadReceipt> CompleteAsync(
bool dryRun, Guid reservation, LeadReceipt receipt, Guid? targetId, CancellationToken ct)
{
Expand Down
4 changes: 3 additions & 1 deletion src/MUI.Web/Mcp/CrawlAdminTools.cs
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,9 @@ public async Task<CrawlLeadAddResult> 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<IReadOnlyList<CrawlLeadRow>> CrawlLeadsAsync(
[Description("How many leads to list. Default 100.")] int batch = 100,
Expand Down
25 changes: 25 additions & 0 deletions tests/MUI.Crawler.Tests/SubmissionPostgresTests.Leads.cs
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,31 @@ await Assert.That(await log.TryBeginAsync(
.IsFalse();
}

/// <summary>
/// Abandoning gives back a bare reservation and never touches a lead that reached an outcome.
/// </summary>
[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]);
}

/// <summary>The bound holds under a concurrent burst, as the form's does.</summary>
[Test]
public async Task AConcurrentBurstOfLeadsDoesNotWalkThroughTheBound()
Expand Down
45 changes: 45 additions & 0 deletions tests/MUI.Discovery.Tests/Intake/LeadServiceTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,51 @@ public async Task TheWindowTakesOnlySoManyLeadsAndThenReopens()
await Assert.That(tomorrow.Outcome).IsEqualTo(SubmissionOutcome.Accepted);
}

/// <summary>
/// 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.
/// </summary>
[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);
}

/// <summary>
/// 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.
/// </summary>
[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);
}

/// <summary>An address that cannot be read is recorded as that, and fabricates no default port.</summary>
[Test]
[Arguments("", 4201)]
Expand Down
19 changes: 19 additions & 0 deletions tests/MUI.Discovery.Tests/Support/InMemoryLeadLog.cs
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@ public Task<bool> TryBeginAsync(
}
}

/// <summary>Makes the next <see cref="CompleteAsync"/> throw, as a dropped connection would.</summary>
public bool FailNextComplete { get; set; }

public Task CompleteAsync(
Guid id,
SubmittedAddress? address,
Expand All @@ -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)
Expand All @@ -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<IReadOnlyList<LeadRecord>> RecentAsync(int limit, CancellationToken ct)
{
lock (_gate)
Expand Down
Loading