From 4cbed63c976a6f2ffba1ab430b6104572dd7476b Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Wed, 30 Sep 2026 19:18:39 -0400 Subject: [PATCH 1/2] fix(integration): stop re-geocoding unchanged addresses and cascading on 429 Firehose updates re-sent unchanged records and geocoded the same address on every update. Nominatim answered with 429, and each 429 fell through to a shorter address variant with another 1.1s wait, so ingest requests ran 5-10s and tripped the SlowAPIResponse alert. - Skip geocoding on update when the address is unchanged and the event already has coordinates - Return null on the first 429 instead of retrying shorter variants - Add a 5s timeout to the Nominatim request --- .../event-integration.service.spec.ts | 63 +++++++++++++++++++ .../services/event-integration.service.ts | 16 ++++- 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/src/event/services/event-integration.service.spec.ts b/src/event/services/event-integration.service.spec.ts index a5f8e0a7..333ceb1d 100644 --- a/src/event/services/event-integration.service.spec.ts +++ b/src/event/services/event-integration.service.spec.ts @@ -19,6 +19,7 @@ import { Counter, Histogram } from 'prom-client'; import { BlueskyIdService } from '../../bluesky/bluesky-id.service'; import { FileService } from '../../file/file.service'; import { EventEmitter2 } from '@nestjs/event-emitter'; +import axios from 'axios'; // Add constants for metrics tokens const PROM_METRIC_EVENT_INTEGRATION_PROCESSED_TOTAL = @@ -584,6 +585,68 @@ describe('EventIntegrationService', () => { ); expect(residualMs).toBeGreaterThanOrEqual(simulatedCreatorIoMs - 1); }); + + it('should not re-geocode an update whose address is unchanged and already geocoded', async () => { + eventQueryService.findBySourceAttributes.mockResolvedValue([ + { + ...mockExistingEvent, + location: '123 Main St, Louisville, KY', + lat: 38.25, + lon: -85.76, + } as unknown as EventEntity, + ]); + const geocodeSpy = jest + .spyOn(service as any, 'geocodeAddress') + .mockResolvedValue(null); + + const result = await service.processExternalEvent( + eventWithAddressOnly, + 'tenant1', + ); + + expect(geocodeSpy).not.toHaveBeenCalled(); + expect(result.lat).toBe(38.25); + expect(result.lon).toBe(-85.76); + }); + + it('should geocode an update whose address changed', async () => { + eventQueryService.findBySourceAttributes.mockResolvedValue([ + { + ...mockExistingEvent, + location: '1 Old Rd, Lexington, KY', + lat: 38.04, + lon: -84.5, + } as unknown as EventEntity, + ]); + const geocodeSpy = jest + .spyOn(service as any, 'geocodeAddress') + .mockResolvedValue({ lat: 38.25, lon: -85.76 }); + + const result = await service.processExternalEvent( + eventWithAddressOnly, + 'tenant1', + ); + + expect(geocodeSpy).toHaveBeenCalledWith('123 Main St, Louisville, KY'); + expect(result.lat).toBe(38.25); + }); + + it('should stop at the first 429 instead of retrying shorter addresses', async () => { + const getSpy = jest.spyOn(axios, 'get').mockRejectedValue( + Object.assign(new Error('Request failed with status code 429'), { + isAxiosError: true, + response: { status: 429 }, + }), + ); + + const coords = await (service as any).geocodeAddress( + 'Boulder, Boulder, CO, US', + ); + + expect(coords).toBeNull(); + expect(getSpy).toHaveBeenCalledTimes(1); + getSpy.mockRestore(); + }); }); describe('Unique violation on insert (23505)', () => { diff --git a/src/event/services/event-integration.service.ts b/src/event/services/event-integration.service.ts index 1ab28dd3..daba1904 100644 --- a/src/event/services/event-integration.service.ts +++ b/src/event/services/event-integration.service.ts @@ -25,6 +25,7 @@ export class EventIntegrationService { private readonly tracer = trace.getTracer('event-integration-service'); private lastGeocodingRequest = 0; private readonly geocodingDelayMs = 1100; // Nominatim rate limit: 1 req/sec, so wait 1.1s to be safe + private readonly geocodingTimeoutMs = 5000; constructor( private readonly tenantService: TenantConnectionService, @@ -458,6 +459,7 @@ export class EventIntegrationService { headers: { 'User-Agent': 'OpenMeet/1.0 (https://openmeet.net)', }, + timeout: this.geocodingTimeoutMs, }, ); @@ -492,6 +494,12 @@ export class EventIntegrationService { `Error geocoding "${currentAddress}": ${error.message}`, ); + // Rate limited: every shorter variant would be refused too, and each + // retry costs another geocodingDelayMs inside the request + if (axios.isAxiosError(error) && error.response?.status === 429) { + return null; + } + // On error, also try with shorter address const commaIndex = currentAddress.indexOf(','); if (commaIndex === -1) { @@ -762,13 +770,19 @@ export class EventIntegrationService { // Handle location if (eventData.location) { + // A re-sent record with the same address keeps its stored coordinates; + // geocoding it again on every update is what got us 429'd by Nominatim + const alreadyGeocoded = + eventData.location.description === existingEvent.location && + !!existingEvent.lat && + !!existingEvent.lon; if (eventData.location.description) { existingEvent.location = eventData.location.description; } if (eventData.location.lat && eventData.location.lon) { existingEvent.lat = eventData.location.lat; existingEvent.lon = eventData.location.lon; - } else if (eventData.location.description) { + } else if (eventData.location.description && !alreadyGeocoded) { // Try to geocode the address if we have a description but no coordinates const coords = await this.geocodeAddress( eventData.location.description, From 9efcb27698d6952aa14a70c6193312f978323526 Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Wed, 30 Sep 2026 22:01:10 -0400 Subject: [PATCH 2/2] fix(pds): treat HandleNotFound as an available handle PDS 0.4.5036 answers resolveHandle for a missing handle with error "HandleNotFound" instead of "InvalidRequest". isHandleAvailable only recognized the old form, so it threw on every available handle and account provisioning never created an atproto identity. Accept both forms so the check works on the current PDS and on older ones. --- src/pds/pds-account.service.spec.ts | 26 ++++++++++++++++++++++++++ src/pds/pds-account.service.ts | 8 ++++++-- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/src/pds/pds-account.service.spec.ts b/src/pds/pds-account.service.spec.ts index a8980f89..a5460980 100644 --- a/src/pds/pds-account.service.spec.ts +++ b/src/pds/pds-account.service.spec.ts @@ -243,6 +243,32 @@ describe('PdsAccountService', () => { ); }); + it('should return true when a newer PDS reports HandleNotFound (400)', async () => { + const errorResponse: AxiosError = { + isAxiosError: true, + response: { + data: { + error: 'HandleNotFound', + message: 'Unable to resolve handle', + }, + status: 400, + statusText: 'Bad Request', + headers: {}, + config: { headers: new AxiosHeaders() }, + }, + message: 'Request failed with status code 400', + name: 'AxiosError', + config: { headers: new AxiosHeaders() }, + toJSON: () => ({}), + }; + + httpService.get.mockReturnValue(throwError(() => errorResponse)); + + const result = await service.isHandleAvailable('available.dev.opnmt.me'); + + expect(result).toBe(true); + }); + it('should return false when handle exists', async () => { const successResponse: AxiosResponse = { data: { diff --git a/src/pds/pds-account.service.ts b/src/pds/pds-account.service.ts index 866b43a1..e73e3763 100644 --- a/src/pds/pds-account.service.ts +++ b/src/pds/pds-account.service.ts @@ -227,9 +227,13 @@ export class PdsAccountService { | { error?: string; message?: string } | undefined; - // AT Protocol returns "InvalidRequest" with message about resolution failure - // when a handle doesn't exist. Other 400 errors (like InvalidHandle for + // PDS 0.4.5036+ returns "HandleNotFound" when a handle doesn't exist; + // older versions return "InvalidRequest" with a message about the + // resolution failure. Other 400 errors (like InvalidHandle for // malformed input) should be thrown. + if (data?.error === 'HandleNotFound') { + return true; + } if ( data?.error === 'InvalidRequest' && data?.message &&