Merge pull request #1846 from bluesky-social/import-send-ticks

Send ticks during import

packages/common-web/src/async.ts

@@ -90,6 +90,10 @@ export class AsyncBuffer<T> {
     return this.buffer.length
   }
 
+  get isClosed(): boolean {
+    return this.closed
+  }
+
   resetPromise() {
     this.promise = new Promise<void>((r) => (this.resolve = r))
   }
@@ -128,6 +132,7 @@ export class AsyncBuffer<T> {
 
   throw(err: unknown) {
     this.toThrow = err
+    this.closed = true
     this.resolve()
   }
 

packages/pds/src/api/com/atproto/temp/importRepo.ts

@@ -4,7 +4,7 @@ import PQueue from 'p-queue'
 import axios from 'axios'
 import { CID } from 'multiformats/cid'
 import { InvalidRequestError } from '@atproto/xrpc-server'
-import { AsyncBuffer, TID } from '@atproto/common'
+import { AsyncBuffer, TID, wait } from '@atproto/common'
 import { AtUri } from '@atproto/syntax'
 import { Repo, WriteOpAction, readCarStream, verifyDiff } from '@atproto/repo'
 import { BlobRef, LexValue } from '@atproto/lexicon'
@@ -22,6 +22,9 @@ export default function (server: Server, ctx: AppContext) {
     handler: async ({ params, input, req }) => {
       const { did } = params
       const outBuffer = new AsyncBuffer<string>()
+      sendTicks(outBuffer).catch((err) => {
+        req.log.error({ err }, 'failed to send ticks')
+      })
       processImport(ctx, did, input.body, outBuffer).catch(async (err) => {
         try {
           await ctx.actorStore.destroy(did)
@@ -39,6 +42,13 @@ export default function (server: Server, ctx: AppContext) {
   })
 }
 
+const sendTicks = async (outBuffer: AsyncBuffer<string>) => {
+  while (!outBuffer.isClosed) {
+    outBuffer.push('tick\n')
+    await wait(1000)
+  }
+}
+
 const processImport = async (
   ctx: AppContext,
   did: string,