Appview: control max thread depth (#3015)

* appview: timeout getFollowsFollowing, fail open

* build

* reduce thread size

* setup thread depth configs

* tidy

.github/workflows/build-and-push-bsky-aws.yaml

@@ -3,7 +3,6 @@ on:
   push:
     branches:
       - main
-      - divy/timeout-polo
 env:
   REGISTRY: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_REGISTRY }}
   USERNAME: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_USERNAME }}

packages/bsky/src/api/app/bsky/feed/getPostThread.ts

@@ -22,6 +22,7 @@ import { HydrateCtx, Hydrator } from '../../../../hydration/hydrator'
 import { Views } from '../../../../views'
 import { DataPlaneClient, isDataplaneError, Code } from '../../../../data-plane'
 import { postUriToThreadgateUri } from '../../../../util/uris'
+import { ServerConfig } from '../../../../config'
 
 export default function (server: Server, ctx: AppContext) {
   const getPostThread = createPipeline(
@@ -75,7 +76,7 @@ const skeleton = async (inputs: SkeletonFnInput<Context, Params>) => {
     const res = await ctx.dataplane.getThread({
       postUri: anchor,
       above: params.parentHeight,
-      below: params.depth,
+      below: getDepth(ctx, anchor, params),
     })
     return {
       anchor,
@@ -109,7 +110,7 @@ const presentation = (
   const { ctx, params, skeleton, hydration } = inputs
   const thread = ctx.views.thread(skeleton, hydration, {
     height: params.parentHeight,
-    depth: params.depth,
+    depth: getDepth(ctx, skeleton.anchor, params),
   })
   if (isNotFoundPost(thread)) {
     // @TODO technically this could be returned as a NotFoundPost based on lexicon
@@ -132,6 +133,7 @@ type Context = {
   dataplane: DataPlaneClient
   hydrator: Hydrator
   views: Views
+  cfg: ServerConfig
 }
 
 type Params = QueryParams & { hydrateCtx: HydrateCtx }
@@ -140,3 +142,11 @@ type Skeleton = {
   anchor: string
   uris: string[]
 }
+
+const getDepth = (ctx: Context, anchor: string, params: Params) => {
+  let maxDepth = ctx.cfg.maxThreadDepth
+  if (ctx.cfg.bigThreadUris.has(anchor) && ctx.cfg.bigThreadDepth) {
+    maxDepth = ctx.cfg.bigThreadDepth
+  }
+  return maxDepth ? Math.min(maxDepth, params.depth) : params.depth
+}

packages/bsky/src/config.ts

@@ -41,6 +41,10 @@ export interface ServerConfigValues {
   blobCacheLocation?: string
   statsigKey?: string
   statsigEnv?: string
+  // threads
+  bigThreadUris: Set<string>
+  bigThreadDepth?: number
+  maxThreadDepth?: number
   // client config
   clientCheckEmailConfirmed?: boolean
 }
@@ -133,6 +137,13 @@ export class ServerConfig {
       !indexedAtEpoch || !isNaN(indexedAtEpoch.getTime()),
       'invalid BSKY_INDEXED_AT_EPOCH',
     )
+    const bigThreadUris = new Set(envList(process.env.BSKY_BIG_THREAD_URIS))
+    const bigThreadDepth = process.env.BSKY_BIG_THREAD_DEPTH
+      ? parseInt(process.env.BSKY_BIG_THREAD_DEPTH || '', 10)
+      : undefined
+    const maxThreadDepth = process.env.BSKY_MAX_THREAD_DEPTH
+      ? parseInt(process.env.BSKY_MAX_THREAD_DEPTH || '', 10)
+      : undefined
     return new ServerConfig({
       version,
       debugMode,
@@ -170,6 +181,9 @@ export class ServerConfig {
       statsigEnv,
       clientCheckEmailConfirmed,
       indexedAtEpoch,
+      bigThreadUris,
+      bigThreadDepth,
+      maxThreadDepth,
       ...stripUndefineds(overrides ?? {}),
     })
   }
@@ -330,6 +344,18 @@ export class ServerConfig {
   get indexedAtEpoch() {
     return this.cfg.indexedAtEpoch
   }
+
+  get bigThreadUris() {
+    return this.cfg.bigThreadUris
+  }
+
+  get bigThreadDepth() {
+    return this.cfg.bigThreadDepth
+  }
+
+  get maxThreadDepth() {
+    return this.cfg.maxThreadDepth
+  }
 }
 
 function stripUndefineds(

packages/dev-env/src/bsky.ts

@@ -64,6 +64,7 @@ export class TestBsky {
       bsyncHttpVersion: '1.1',
       modServiceDid: cfg.modServiceDid ?? 'did:example:invalidMod',
       labelsFromIssuerDids: [EXAMPLE_LABELER],
+      bigThreadUris: new Set(),
       ...cfg,
       adminPasswords: [ADMIN_PASSWORD],
     })