# Thunder GetInNetworkPosts

> InNetworkPostsService RPC fields, PostStore timelines and retention, Kafka ingest, semaphore capacity, and Strato following-list fallback.

- Repository: xai-org/x-algorithm
- GitHub: https://github.com/xai-org/x-algorithm
- Human docs: https://grok-wiki.com/public/docs/xai-org-x-algorithm-23c09c39074c
- Complete Markdown: https://grok-wiki.com/public/docs/xai-org-x-algorithm-23c09c39074c/llms-full.txt

## Source Files

- `thunder/thunder_service.rs`
- `thunder/posts/post_store.rs`
- `thunder/main.rs`
- `thunder/kafka_utils.rs`
- `thunder/kafka/tweet_events_listener.rs`
- `home-mixer/sources/thunder_source.rs`
- `thunder/lib.rs`

---

---
title: "Thunder GetInNetworkPosts"
description: "InNetworkPostsService RPC fields, PostStore timelines and retention, Kafka ingest, semaphore capacity, and Strato following-list fallback."
---

`InNetworkPostsService.get_in_network_posts` is Thunder's in-network candidate RPC. `ThunderServiceImpl` acquires a process-wide `tokio::sync::Semaphore` with `try_acquire`, hydrates an empty following list from `StratoClient` only when `debug` is also true, scans in-memory `PostStore` timelines on a blocking thread, then returns the newest `LightPost`s up to `max_results`. Home Mixer calls this through `ThunderSource`. The `thunder/` tree in this checkout is a production snapshot: `args`, `config`, `strato_client`, `metrics`, `schema`, `o2`, `xai_thunder_proto`, `xai_kafka`, and `xai_http_server` are not published here, so the binary does not build or serve locally.

<Warning>
This checkout cannot start Thunder. Topic and destination strings are empty, SASL env-var names are empty, and numeric caps such as `MAX_POSTS_TO_RETURN` live in the unpublished `thunder/config` module. Use this page as the production contract, not as a local runbook.
</Warning>

## Architecture

```mermaid
flowchart LR
  subgraph Feeder["Thunder feeder — is_serving = false"]
    TE["Thrift TweetEvent Kafka"] --> L1["tweet_events_listener"]
    L1 --> PR["InNetworkEvent producer"]
  end

  subgraph Serving["Thunder serving — is_serving = true"]
    PR --> V2["tweet_events_listener_v2"]
    V2 --> PS["PostStore DashMaps"]
    STR["StratoClient"] -.->|"empty following AND debug"| SVC
    PS --> SVC["ThunderServiceImpl"]
    SVC --> SEM["request_semaphore"]
  end

  subgraph HomeMixer["Home Mixer"]
    FUH["FollowedUserIdsQueryHydrator"] --> TS["ThunderSource"]
    TS -->|"GetInNetworkPostsRequest<br/>debug = false"| SVC
    SVC --> TS
    TS --> PC["PostCandidate + InNetworkReply"]
  end
```

Two processes share the same crate. The feeder (`!args.is_serving`) consumes Thrift `TweetEvent`s and publishes protobuf `InNetworkEvent`s. Serving Thunder (`args.is_serving`) consumes those events into `PostStore`, then exposes `InNetworkPostsService` over HTTP/gRPC via `xai_http_server`.

## Startup

`thunder/main.rs` always constructs `PostStore`, `StratoClient`, and `ThunderServiceImpl`, then starts Kafka. Serving mode blocks readiness on Kafka catch-up.

<Steps>
  <Step title="Construct store and service">
    `PostStore::new(args.post_retention_seconds, args.request_timeout_ms)`. `ThunderServiceImpl::new` takes that store, a `StratoClient`, and `args.max_concurrent_requests` as the semaphore permit count. The gRPC server is `InNetworkPostsServiceServer` with Zstd accept and send compression.
  </Step>
  <Step title="Start Kafka">
    `kafka_utils::start_kafka` opens an `mpsc` channel of size `args.kafka_num_threads`. Serving mode starts `start_tweet_event_processing_v2`. Feeder mode starts `start_tweet_event_processing` plus an `InNetworkEvent` producer.
  </Step>
  <Step title="Serving catch-up">
    Serving Thunder waits for one channel message per Kafka thread, then `post_store.finalize_init()` (sort timelines, trim, re-drop tombstoned IDs). It starts the 5-second stats logger and a 2-minute auto-trim. Only then does `http_server.set_readiness(true)`.
  </Step>
</Steps>

`PostStore::default()` is 2 days of retention and a 0 ms request timeout (no timeout). Production serving uses the CLI values, not that default.

## RPC: GetInNetworkPosts

:::endpoint POST InNetworkPostsService/GetInNetworkPosts Recent posts from followed authors
Thunder implements `InNetworkPostsService`. Home Mixer builds `GetInNetworkPostsRequest` in `ThunderSource`; the handler is `ThunderServiceImpl::get_in_network_posts`.

<ParamField body="user_id" type="u64" required>
Viewer ID. Used for Strato fallback, request-timeout logs, and dropping retweets whose `source_user_id` is the viewer.
</ParamField>

<ParamField body="following_user_ids" type="repeated u64">
Author IDs to scan. Home Mixer copies `query.user_features.followed_user_ids`. Truncated to the first `MAX_INPUT_LIST_SIZE` entries.
</ParamField>

<ParamField body="exclude_tweet_ids" type="repeated u64">
Post IDs to skip. Home Mixer sends `query.seen_ids`. Truncated to `MAX_INPUT_LIST_SIZE`.
</ParamField>

<ParamField body="max_results" type="u32">
Cap after recency sort. `0` means `MAX_VIDEOS_TO_RETURN` when `is_video_request`, otherwise `MAX_POSTS_TO_RETURN`. Those constants are unpublished.
</ParamField>

<ParamField body="algorithm" type="param">
Home Mixer sets `query.params.get(ThunderAlgorithm)`. The handler never reads this field.
</ParamField>

<ParamField body="debug" type="bool">
Enables request/response logs. Also required for Strato fallback. `ThunderSource` always sends `false`.
</ParamField>

<ParamField body="is_video_request" type="bool">
Selects `get_videos_by_users` instead of `get_all_posts_by_users`. `ThunderSource` always sends `false`.
</ParamField>

<ResponseField name="posts" type="repeated LightPost">
Recency-sorted posts after store filters and `score_recent`.
</ResponseField>
:::

<RequestExample>
```text
GetInNetworkPostsRequest  # Home Mixer ThunderSource
user_id:             query.user_id
following_user_ids:  query.user_features.followed_user_ids as u64
max_results:         query.params.get(ThunderMaxResults)
exclude_tweet_ids:   query.seen_ids
algorithm:           query.params.get(ThunderAlgorithm)
debug:               false
is_video_request:    false
```
</RequestExample>

<ResponseExample>
```text
GetInNetworkPostsResponse
posts[] LightPost:
  post_id, author_id, created_at
  in_reply_to_post_id, in_reply_to_user_id
  is_retweet, is_reply
  source_post_id, source_user_id
  has_video, conversation_id
```
</ResponseExample>

Handler sequence:

1. `request_semaphore.try_acquire()`. Failure increments `REJECTED_REQUESTS` and returns immediately.
2. If `following_user_ids` is empty **and** `debug`, call `strato_client.fetch_following_list(user_id, MAX_INPUT_LIST_SIZE)`.
3. Truncate following and exclude lists. Resolve `max_results`.
4. `tokio::task::spawn_blocking`: `get_videos_by_users` or `get_all_posts_by_users`, then `score_recent`.
5. Return `GetInNetworkPostsResponse { posts }`.

`score_recent` sorts by `created_at` descending (`sort_unstable_by_key(Reverse)`) and `take(max_results)`. There is no engagement score inside Thunder.

```mermaid
sequenceDiagram
  participant HM as ThunderSource
  participant S as ThunderServiceImpl
  participant Sem as request_semaphore
  participant St as StratoClient
  participant PS as PostStore

  HM->>S: GetInNetworkPostsRequest
  S->>Sem: try_acquire
  alt no permit
    Sem-->>S: RESOURCE_EXHAUSTED
    S-->>HM: Status resource_exhausted
  else permit
    opt following empty AND debug
      S->>St: fetch_following_list
      St-->>S: user IDs or INTERNAL
    end
    S->>PS: get_all_posts_by_users / get_videos_by_users
    PS-->>S: LightPost[]
    S-->>HM: GetInNetworkPostsResponse
  end
```

## Semaphore capacity

The permit count is `args.max_concurrent_requests`. The handler never waits.

| Condition | Metric | gRPC status | Message |
|-----------|--------|-------------|---------|
| `try_acquire` fails | `REJECTED_REQUESTS` | `RESOURCE_EXHAUSTED` | `Server at capacity, please retry` |
| Permit granted | `IN_FLIGHT_REQUESTS` +1, Drop guard −1 | continues | — |
| Strato fetch fails | — | `INTERNAL` | `Failed to fetch following list: …` |
| `spawn_blocking` join fails | — | `INTERNAL` | `Failed to process posts: …` |

<Warning>
Home Mixer maps any Thunder status to `Err(format!("ThunderSource: {}", e))`. A capacity reject surfaces as that string, not as a retry inside `ThunderSource`.
</Warning>

A second semaphore exists only on the serving ingest path: `Semaphore::new(3)` in `tweet_events_listener_v2`. After Kafka catch-up, each insert/delete batch acquires one of those three permits so ingest does not starve RPC CPU.

## Strato following-list fallback

```text
if req.following_user_ids.is_empty() && req.debug:
    strato_client.fetch_following_list(user_id as i64, MAX_INPUT_LIST_SIZE as i32)
```

| Caller | `following_user_ids` | `debug` | Strato? |
|--------|----------------------|---------|---------|
| `ThunderSource` | `FollowedUserIdsQueryHydrator` / SocialGraph | `false` | Never |
| Debug RPC with a non-empty list | provided | `true` | No |
| Debug RPC with an empty list | empty | `true` | Yes; failure is `INTERNAL` |
| Non-debug RPC with an empty list | empty | `false` | No; store scan runs with zero authors |

Home Mixer therefore never uses Thunder's Strato path. An empty SocialGraph following list yields an empty Thunder response, not a fallback fetch.

## PostStore timelines and retention

`PostStore` keeps full `LightPost`s in `posts: DashMap<i64, LightPost>` and three per-author `VecDeque<TinyPost>` timelines (`post_id` + `created_at` only).

| Map | Membership | Per-author take |
|-----|------------|-----------------|
| `original_posts_by_user` | `!is_reply && !is_retweet` | `MAX_ORIGINAL_POSTS_PER_AUTHOR` |
| `secondary_posts_by_user` | replies and retweets | `MAX_REPLY_POSTS_PER_AUTHOR` |
| `video_posts_by_user` | video-eligible posts | `MAX_VIDEO_POSTS_PER_AUTHOR` |
| `deleted_posts` | tombstones | not scanned as a timeline |

**Insert.** `insert_posts` drops posts with `created_at >= now` or age `> retention_seconds`, sorts remaining by `created_at`, then `insert_posts_internal`. Already-tombstoned IDs and already-stored IDs are skipped.

**Video eligibility** (store insert):

- Replies are never video-eligible.
- Otherwise `has_video`, or a retweet whose `source_post_id` is a non-reply `LightPost` with `has_video`.

Feeder ingest sets `has_video` only when the first media entity is `VideoInfo` and `duration_millis >= MIN_VIDEO_DURATION_MS`.

**Deletes.** `mark_as_deleted` removes the `LightPost`, inserts a tombstone, and appends a `TinyPost` under `DELETE_EVENT_KEY` on `original_posts_by_user` so tombstones expire with the same trim loop.

**Lookup.** `get_posts_from_map` walks authors in request order. Per author it:

1. Breaks the whole request if `request_timeout` is non-zero and elapsed (`POST_STORE_REQUEST_TIMEOUTS`).
2. Scans the deque newest-first, skips `exclude_tweet_ids`, and stops after `MAX_TINY_POSTS_PER_USER_SCAN`.
3. Resolves `LightPost`, drops tombstones, drops retweets with `source_user_id == request_user_id`.
4. For the secondary map only (`following_users` non-empty), keeps a reply when:
   - `in_reply_to_post_id` is unset, or
   - the replied-to post is an original (not reply, not retweet), or
   - it is a reply-to-reply whose `in_reply_to_post_id` equals `conversation_id` **and** `in_reply_to_user_id` is in the following set.
   Missing parent posts drop the reply.
5. Takes at most `max_per_user`.

`get_all_posts_by_users` concatenates original + secondary. `get_videos_by_users` reads only `video_posts_by_user`.

**Retention.** Serving auto-trim runs every 2 minutes. `trim_old_posts` pops deque fronts older than `retention_seconds`, removes matching `posts` entries, and expires `DELETE_EVENT_KEY` tombstones. `finalize_init` sorts every deque by `created_at` ascending, trims, then removes any ID still listed in `deleted_posts` so out-of-order create/delete during catch-up cannot resurrect a post.

## Kafka ingest

Topic and dest constants in `kafka_utils.rs` are empty strings in this snapshot. The mode split is still in code.

| Mode | `args.is_serving` | Consumer | Writer |
|------|-------------------|----------|--------|
| Feeder | `false` | Thrift `TweetEvent` (`deserialize_tweet_event`) | `InNetworkEvent` producer; does **not** write `PostStore` |
| Serving | `true` | Protobuf `InNetworkEvent` (`deserialize_tweet_event_v2`) | `PostStore::insert_posts` / `mark_as_deleted` |

**Feeder filters**

- Skip `TweetCreateEvent` when `core_data.nullcast` is true.
- Skip `TweetDeleteEvent` when `now - created_at_secs > post_retention_seconds`.
- Treat `QuotedTweetDeleteEvent` as a delete of `quoting_tweet_id`.
- Other event variants are logged and ignored.

**Serving consumer**

- `group_id` is `{kafka_group_id}-{uuid}`.
- `max_partition_fetch_bytes` is 100 MiB (feeder uses 10 MiB).
- Threads split `args.kafka_tweet_events_v2_num_partitions`.
- Catch-up: when `sum(partition lag) < lags.len() * kafka_batch_size`, the thread sends on the init channel. `main` waits for `kafka_num_threads` sends.
- After catch-up, insert batches hold one of three ingest permits.
- v2 marks `is_reply` if the flag is set **or** `in_reply_to_post_id` / `in_reply_to_user_id` is present.

Both modes spawn `start_partition_lag_monitor` on `args.lag_monitor_interval_secs`. Poll errors increment `KAFKA_POLL_ERRORS` and sleep 100 ms.

## Home Mixer caller

`PhoenixCandidatePipeline` always includes `ThunderSource` next to Phoenix, Tweet Mixer, and `CachedPostsSource`.

| Item | Behavior |
|------|----------|
| `enable` | `!query.has_cached_posts` (cache hit of ≥ 500 Redis posts disables Thunder) |
| Channel | `ThunderCluster::parse(ThunderClusterId)` then `ThunderCluster::resolve(..., decider)`; missing channel is `"ThunderSource: no available channel"` |
| Following | `query.user_features.followed_user_ids` from `FollowedUserIdsQueryHydrator` |
| Exclude | `query.seen_ids` |
| `served_type` | `ForYouInNetwork` unless `query.in_network_only`, then `RankedFollowing` |
| Replies | Each post with `in_reply_to_post_id` is stored on `query.in_network_replies` |
| Ancestors | `[in_reply_to_post_id]` plus `conversation_id` when it differs |

`InNetworkCandidateHydrator` later sets `candidate.in_network` from the same following list (or self). That flag is independent of Thunder's store filters.

## Args used by this snapshot

`thunder/args` is unpublished. These fields are read from `Args`:

| Field | Used for |
|-------|----------|
| `post_retention_seconds` | Store retention, insert filter, feeder delete skip, log days |
| `request_timeout_ms` | Mid-scan abort in `get_posts_from_map` |
| `max_concurrent_requests` | RPC semaphore permits |
| `grpc_port`, `http_port` | `HttpServer` + `GrpcConfig` |
| `enable_profiling` | `xai_profiling::spawn_server(3000, …)` |
| `is_serving` | Feeder vs serving Kafka path and catch-up |
| `kafka_num_threads` | Thread count and catch-up barrier |
| `kafka_batch_size` | Batch threshold and catch-up lag compare |
| `kafka_group_id` | Consumer group prefix |
| `kafka_tweet_events_v2_num_partitions` | Serving partition range |
| `tweet_events_num_partitions` | Feeder partition range |
| `in_network_events_consumer_dest` | Serving consumer dest |
| `auto_offset_reset`, `skip_to_latest`, `fetch_timeout_ms` | Consumer config |
| `lag_monitor_interval_secs` | Partition-lag scrape |
| `security_protocol`, SASL user/mechanism/password (consumer and producer) | Kafka SSL |

Optional profiling is independent of readiness. Termination is `http_server.wait_for_termination()`.

## Unpublished config identifiers

Referenced but not defined in this checkout:

`MAX_INPUT_LIST_SIZE`, `MAX_POSTS_TO_RETURN`, `MAX_VIDEOS_TO_RETURN`, `MAX_ORIGINAL_POSTS_PER_AUTHOR`, `MAX_REPLY_POSTS_PER_AUTHOR`, `MAX_VIDEO_POSTS_PER_AUTHOR`, `MAX_TINY_POSTS_PER_USER_SCAN`, `DELETE_EVENT_KEY`, `MIN_VIDEO_DURATION_MS`.

Do not invent values. The only numeric defaults present in-repo are `PostStore::default()` (2 days, 0 ms timeout), ingest semaphore `3`, auto-trim interval `2` minutes, stats interval `5` seconds, and cached-posts disable threshold `500`.

## Errors and verification

| Symptom | Cause | Check |
|---------|--------|--------|
| `ThunderSource: status: ResourceExhausted, message: "Server at capacity, please retry"` | RPC semaphore exhausted | Lower `max_concurrent_requests` pressure or retry; Thunder does not queue |
| `ThunderSource: no available channel` | `ThunderClient.get_random_channel` returned `None` | Cluster id / decider, not PostStore |
| Empty in-network set with a populated following list | Retention miss, excludes, timeout mid-scan, or reply/retweet filters | `POST_STORE_REQUEST_TIMEOUTS`, `POST_STORE_POSTS_RETURNED` |
| Empty set and empty following | SocialGraph returned no IDs; Strato does not run because `debug` is false | `FollowedUserIdsQueryHydrator` |
| Thunder never called | `has_cached_posts` | Redis cache ≥ 500 candidates |
| Binary will not build | Unpublished crates and empty Kafka dests | Expected in this checkout |

Serving logs `Kafka init took …` then `HTTP/gRPC server is ready`. Stats lines look like `PostStore Stats: N users, M total posts, D deleted posts`.

## Next

<CardGroup>
  <Card title="In-network and out-of-network" href="/in-network-out-of-network">
    Thunder versus Phoenix enable predicates and how `in_network_only` changes `served_type`.
  </Card>
  <Card title="Candidate sources" href="/candidate-sources">
    Where `ThunderSource` sits in `PhoenixCandidatePipeline` next to Phoenix and cached posts.
  </Card>
  <Card title="Runtime boundaries" href="/runtime-boundaries">
    What this checkout can execute versus unpublished Thunder and Home Mixer crates.
  </Card>
  <Card title="Troubleshooting" href="/troubleshooting">
    `RESOURCE_EXHAUSTED` and other For You failure signals.
  </Card>
</CardGroup>