courier push retries

packages/bsky/src/notifications.ts

@@ -11,7 +11,7 @@ import { NotificationPushToken as PushToken } from './db/tables/notification-pus
 import logger from './indexer/logger'
 import { notSoftDeletedClause, valuesList } from './db/util'
 import { ids } from './lexicon/lexicons'
-import { retryHttp } from './util/retry'
+import { retryConnect, retryHttp } from './util/retry'
 import { Notification as CourierNotification } from './proto/courier_pb'
 import { CourierClient } from './courier'
 
@@ -339,8 +339,8 @@ export class CourierNotificationServer extends NotificationServer<CourierNotific
         alwaysDeliver: !n.rateLimit,
         timestamp: Timestamp.fromDate(new Date(n.notif.sortAt)),
         additional: Struct.fromJson({
-          uri: n.notif.recordUri || '',
-          reason: n.notif.reason || '',
+          uri: n.notif.recordUri,
+          reason: n.notif.reason,
           subject: n.notif.reasonSubject || '',
         }),
       })
@@ -350,7 +350,9 @@ export class CourierNotificationServer extends NotificationServer<CourierNotific
 
   async processNotifications(prepared: CourierNotification[]): Promise<void> {
     try {
-      await this.courierClient.pushNotifications({ notifications: prepared })
+      await retryConnect(() =>
+        this.courierClient.pushNotifications({ notifications: prepared }),
+      )
     } catch (err) {
       logger.error({ err }, 'notification push to courier failed')
     }

packages/bsky/src/util/retry.ts

@@ -1,6 +1,7 @@
 import { AxiosError } from 'axios'
 import { XRPCError, ResponseType } from '@atproto/xrpc'
 import { RetryOptions, retry } from '@atproto/common'
+import { Code, ConnectError } from '@connectrpc/connect'
 
 export async function retryHttp<T>(
   fn: () => Promise<T>,
@@ -24,3 +25,14 @@ export function retryableHttp(err: unknown) {
 const retryableHttpStatusCodes = new Set([
   408, 425, 429, 500, 502, 503, 504, 522, 524,
 ])
+
+export async function retryConnect<T>(
+  fn: () => Promise<T>,
+  opts: RetryOptions = {},
+): Promise<T> {
+  return retry(fn, { retryable: retryableConnect, ...opts })
+}
+
+export function retryableConnect(err: unknown) {
+  return err instanceof ConnectError && err.code === Code.Unavailable
+}