Repository: PostHog/posthog
Pull request: feat(flags): use redis reader and writer
> [!IMPORTANT]
> 👉 _Stay up-to-date with [PostHog coding conventions](https://posthog.com/docs/contribute/coding-conventions) for a smoother review._

## Problem

Closes https://github.com/PostHog/posthog/issues/34067

## Changes

- uses separate reader and writer clients for accessing redis, with safe fallbacks so that if the URLs for the various services aren't provided, they fall back to the writer.
- clearly delineate between read operations and write operations, with metrics on each.

## Did you write or update any docs for this change?

- [x] No docs needed for this change

## How did you test this code?

TODO, I feel like this is a thing I can test in prod but maybe I can test some metrics patterns locally or something.


Problem statement:
split out redis reader and writer instances for `/flags`
We overuse the redis writers and don't hit the readers as much, and we've also noticed that under heavy load the that redis sometimes times out.  Our service is read-heavy (although we still write a decent amount for billing), so we're a good candidate for splitting these services out.

Hints from issue discussion:

Commit message: reader and writer now

Patch to review:
diff --git a/rust/feature-flags/src/flags/flag_request.rs b/rust/feature-flags/src/flags/flag_request.rs
index 22e4083f33ee4..d0af4131e42ee 100644
--- a/rust/feature-flags/src/flags/flag_request.rs
+++ b/rust/feature-flags/src/flags/flag_request.rs
@@ -337,7 +337,7 @@ mod tests {
             .extract_token()
             .expect("failed to extract token");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
 
         match flag_service.verify_token(&token).await {
             Ok(extracted_token) => assert_eq!(extracted_token, team.api_token),
@@ -359,7 +359,7 @@ mod tests {
             .extract_token()
             .expect("failed to extract token");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
         assert!(matches!(
             flag_service.verify_token(&result).await,
             Err(FlagError::TokenValidationError)
diff --git a/rust/feature-flags/src/flags/flag_service.rs b/rust/feature-flags/src/flags/flag_service.rs
index 0847406fcbb2c..33a22d0880895 100644
--- a/rust/feature-flags/src/flags/flag_service.rs
+++ b/rust/feature-flags/src/flags/flag_service.rs
@@ -15,17 +15,20 @@ use std::sync::Arc;
 
 /// Service layer for handling feature flag operations
 pub struct FlagService {
-    redis_client: Arc<dyn RedisClient + Send + Sync>,
+    redis_reader: Arc<dyn RedisClient + Send + Sync>,
+    redis_writer: Arc<dyn RedisClient + Send + Sync>,
     pg_client: Arc<dyn DatabaseClient + Send + Sync>,
 }
 
 impl FlagService {
     pub fn new(
-        redis_client: Arc<dyn RedisClient + Send + Sync>,
+        redis_reader: Arc<dyn RedisClient + Send + Sync>,
+        redis_writer: Arc<dyn RedisClient + Send + Sync>,
         pg_client: Arc<dyn DatabaseClient + Send + Sync>,
     ) -> Self {
         Self {
-            redis_client,
+            redis_reader,
+            redis_writer,
             pg_client,
         }
     }
@@ -34,7 +37,7 @@ impl FlagService {
     /// If the token is not found in the cache, it will be verified against the database,
     /// and the result will be cached in redis.
     pub async fn verify_token(&self, token: &str) -> Result<String, FlagError> {
-        let (result, cache_hit) = match Team::from_redis(self.redis_client.clone(), token).await {
+        let (result, cache_hit) = match Team::from_redis(self.redis_reader.clone(), token).await {
             Ok(_) => (Ok(token), true),
             Err(_) => {
                 match Team::from_pg(self.pg_client.clone(), token).await {
@@ -42,7 +45,7 @@ impl FlagService {
                         inc(DB_TEAM_READS_COUNTER, &[], 1);
                         // Token found in PostgreSQL, update Redis cache so that we can verify it from Redis next time
                         if let Err(e) =
-                            Team::update_redis_cache(self.redis_client.clone(), &team).await
+                            Team::update_redis_cache(self.redis_writer.clone(), &team).await
                         {
                             tracing::warn!("Failed to update Redis cache: {}", e);
                             inc(
@@ -79,7 +82,7 @@ impl FlagService {
     /// If the team is not found in the cache, it will be fetched from the database and stored in the cache.
     /// Returns the team if found, otherwise an error.
     pub async fn get_team_from_cache_or_pg(&self, token: &str) -> Result<Team, FlagError> {
-        let (team_result, cache_hit) = match Team::from_redis(self.redis_client.clone(), token)
+        let (team_result, cache_hit) = match Team::from_redis(self.redis_reader.clone(), token)
             .await
         {
             Ok(team) => (Ok(team), true),
@@ -87,7 +90,7 @@ impl FlagService {
                 Ok(team) => {
                     inc(DB_TEAM_READS_COUNTER, &[], 1);
                     // If we have the team in postgres, but not redis, update redis so we're faster next time
-                    if (Team::update_redis_cache(self.redis_client.clone(), &team).await).is_err() {
+                    if (Team::update_redis_cache(self.redis_writer.clone(), &team).await).is_err() {
                         inc(
                             TEAM_CACHE_ERRORS_COUNTER,
                             &[("reason".to_string(), "redis_update_failed".to_string())],
@@ -118,14 +121,14 @@ impl FlagService {
         project_id: i64,
     ) -> Result<FeatureFlagList, FlagError> {
         let (flags_result, cache_hit) =
-            match FeatureFlagList::from_redis(self.redis_client.clone(), project_id).await {
+            match FeatureFlagList::from_redis(self.redis_reader.clone(), project_id).await {
                 Ok(flags) => (Ok(flags), true),
                 Err(_) => {
                     match FeatureFlagList::from_pg(self.pg_client.clone(), project_id).await {
                         Ok(flags) => {
                             inc(DB_FLAG_READS_COUNTER, &[], 1);
                             if (FeatureFlagList::update_flags_in_redis(
-                                self.redis_client.clone(),
+                                self.redis_writer.clone(),
                                 project_id,
                                 &flags,
                             )
@@ -178,7 +181,7 @@ mod tests {
             .await
             .expect("Failed to insert new team in Redis");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
 
         // Test valid token in Redis
         let result = flag_service.verify_token(&team.api_token).await;
@@ -213,7 +216,7 @@ mod tests {
             .await
             .expect("Failed to insert new team in Redis");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
 
         // Test fetching from Redis
         let result = flag_service
@@ -229,7 +232,7 @@ mod tests {
             .await
             .expect("Failed to remove team from Redis");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
 
         let result = flag_service
             .get_team_from_cache_or_pg(&team.api_token)
@@ -336,7 +339,7 @@ mod tests {
             .await
             .expect("Failed to insert mock flags in Redis");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
 
         // Test fetching from Redis
         let result = flag_service
diff --git a/rust/feature-flags/src/handler/billing.rs b/rust/feature-flags/src/handler/billing.rs
index 1215711fd5161..0566fd4253b83 100644
--- a/rust/feature-flags/src/handler/billing.rs
+++ b/rust/feature-flags/src/handler/billing.rs
@@ -50,7 +50,7 @@ pub async fn record_usage(
 
     if has_billable_flags {
         if let Err(e) = increment_request_count(
-            context.state.redis.clone(),
+            context.state.redis_writer.clone(),
             team_id,
             1,
             FlagRequestType::Decide,
diff --git a/rust/feature-flags/src/handler/config_response_builder.rs b/rust/feature-flags/src/handler/config_response_builder.rs
index 9e1a18870bc67..7c9ee12acaaa8 100644
--- a/rust/feature-flags/src/handler/config_response_builder.rs
+++ b/rust/feature-flags/src/handler/config_response_builder.rs
@@ -27,7 +27,7 @@ impl ConfigContext {
         Self {
             config: context.state.config.clone(),
             reader: context.state.reader.clone(),
-            redis: context.state.redis.clone(),
+            redis: context.state.redis_reader.clone(),
             headers: context.headers.clone(),
         }
     }
diff --git a/rust/feature-flags/src/handler/mod.rs b/rust/feature-flags/src/handler/mod.rs
index de8f4a62e9f39..09fe6ba44799e 100644
--- a/rust/feature-flags/src/handler/mod.rs
+++ b/rust/feature-flags/src/handler/mod.rs
@@ -28,8 +28,11 @@ pub async fn process_request(context: RequestContext) -> Result<FlagsResponse, F
     async move {
         let start_time = std::time::Instant::now();
 
-        let flag_service =
-            FlagService::new(context.state.redis.clone(), context.state.reader.clone());
+        let flag_service = FlagService::new(
+            context.state.redis_reader.clone(),
+            context.state.redis_writer.clone(),
+            context.state.reader.clone(),
+        );
 
         let (original_distinct_id, verified_token, request) =
             authentication::parse_and_authenticate(&context, &flag_service).await?;
@@ -70,7 +73,7 @@ pub async fn process_request(context: RequestContext) -> Result<FlagsResponse, F
 
         let property_overrides = properties::prepare_overrides(&context, &request)?;
 
-        // Evaluate flags (this will return empty if disable_flags is true)
+        // Evaluate flags (this will return empty if is_flags_disabled is true)
         let flags_response = flags::evaluate_for_request(
             &context.state,
             team.id,
diff --git a/rust/feature-flags/src/handler/tests.rs b/rust/feature-flags/src/handler/tests.rs
index 9489048415274..2d6ce9172d7a8 100644
--- a/rust/feature-flags/src/handler/tests.rs
+++ b/rust/feature-flags/src/handler/tests.rs
@@ -1097,7 +1097,7 @@ fn test_decode_request_content_types() {
 async fn test_fetch_and_filter_flags() {
     let redis_client = setup_redis_client(None);
     let reader: Arc<dyn Client + Send + Sync> = setup_pg_reader_client(None).await;
-    let flag_service = FlagService::new(redis_client.clone(), reader.clone());
+    let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), reader.clone());
     let team = insert_new_team_in_pg(reader.clone(), None).await.unwrap();
 
     // Create a mix of survey and non-survey flags
diff --git a/rust/feature-flags/src/router.rs b/rust/feature-flags/src/router.rs
index ca4857a30a61f..a19bd054cc6fd 100644
--- a/rust/feature-flags/src/router.rs
+++ b/rust/feature-flags/src/router.rs
@@ -27,7 +27,8 @@ use crate::{
 
 #[derive(Clone)]
 pub struct State {
-    pub redis: Arc<dyn RedisClient + Send + Sync>,
+    pub redis_reader: Arc<dyn RedisClient + Send + Sync>,
+    pub redis_writer: Arc<dyn RedisClient + Send + Sync>,
     pub reader: Arc<dyn DatabaseClient + Send + Sync>,
     pub writer: Arc<dyn DatabaseClient + Send + Sync>,
     pub cohort_cache_manager: Arc<CohortCacheManager>,
@@ -39,8 +40,9 @@ pub struct State {
 }
 
 #[allow(clippy::too_many_arguments)]
-pub fn router<R, D>(
-    redis: Arc<R>,
+pub fn router<RR, RW, D>(
+    redis_reader: Arc<RR>,
+    redis_writer: Arc<RW>,
     reader: Arc<D>,
     writer: Arc<D>,
     cohort_cache: Arc<CohortCacheManager>,
@@ -51,11 +53,13 @@ pub fn router<R, D>(
     config: Config,
 ) -> Router
 where
-    R: RedisClient + Send + Sync + 'static,
+    RR: RedisClient + Send + Sync + 'static,
+    RW: RedisClient + Send + Sync + 'static,
     D: DatabaseClient + Send + Sync + 'static,
 {
     let state = State {
-        redis,
+        redis_reader,
+        redis_writer,
         reader,
         writer,
         cohort_cache_manager: cohort_cache,
diff --git a/rust/feature-flags/src/server.rs b/rust/feature-flags/src/server.rs
index eab573af09c09..c990f8cae4f76 100644
--- a/rust/feature-flags/src/server.rs
+++ b/rust/feature-flags/src/server.rs
@@ -21,6 +21,7 @@ where
     F: Future<Output = ()> + Send + 'static,
 {
     // Create separate Redis clients for reading and writing
+    // NB: if either of these URLs don't exist in the config, we default to the writer
     let redis_reader_client = match RedisClient::new(config.get_redis_reader_url().to_string()) {
         Ok(client) => Arc::new(client),
         Err(e) => {
@@ -33,7 +34,7 @@ where
         }
     };
 
-    let _redis_writer_client = match RedisClient::new(config.get_redis_writer_url().to_string()) {
+    let redis_writer_client = match RedisClient::new(config.get_redis_writer_url().to_string()) {
         Ok(client) => Arc::new(client),
         Err(e) => {
             tracing::error!(
@@ -45,8 +46,7 @@ where
         }
     };
 
-    // For backwards compatibility, use the reader client as the primary Redis client
-    // for services that don't distinguish between reads and writes yet
+    // For backwards compatibility, use the reader client for non-write operations
     let redis_client = redis_reader_client.clone();
 
     let reader = match get_pool(&config.read_database_url, config.max_pg_connections).await {
@@ -118,7 +118,7 @@ where
 
     let billing_limiter = match RedisLimiter::new(
         Duration::from_secs(5),
-        redis_client.clone(),
+        redis_reader_client.clone(), // NB: the limiter only reads from redis, so it's safe to just use the reader client
         QUOTA_LIMITER_CACHE_KEY.to_string(),
         None,
         QuotaResource::FeatureFlags,
@@ -131,14 +131,14 @@ where
         }
     };
 
-    // You can decide which client to pass to the router, or pass both if needed
     let cookieless_manager = Arc::new(CookielessManager::new(
         config.get_cookieless_config(),
-        redis_client.clone(),
+        redis_client.clone(), // NB: the cookieless manager only reads from redis, so it's safe to just use the reader client
     ));
 
     let app = router::router(
-        redis_client,
+        redis_reader_client,
+        redis_writer_client,
         reader,
         writer,
         cohort_cache,
diff --git a/rust/feature-flags/tests/common/mod.rs b/rust/feature-flags/tests/common/mod.rs
index cac3d5a4dcf5b..4a7dbb703ee37 100644
--- a/rust/feature-flags/tests/common/mod.rs
+++ b/rust/feature-flags/tests/common/mod.rs
@@ -76,7 +76,8 @@ impl ServerHandle {
         }
 
         tokio::spawn(async move {
-            let redis_client = Arc::new(mock_client);
+            let redis_reader_client = Arc::new(mock_client);
+            let redis_writer_client = redis_reader_client.clone();
             let reader = match get_pool(&config.read_database_url, config.max_pg_connections).await
             {
                 Ok(client) => Arc::new(client),
@@ -116,7 +117,7 @@ impl ServerHandle {
 
             let billing_limiter = RedisLimiter::new(
                 Duration::from_secs(5),
-                redis_client.clone(),
+                redis_reader_client.clone(),
                 QUOTA_LIMITER_CACHE_KEY.to_string(),
                 None,
                 QuotaResource::FeatureFlags,
@@ -126,11 +127,12 @@ impl ServerHandle {
 
             let cookieless_manager = Arc::new(common_cookieless::CookielessManager::new(
                 config.get_cookieless_config(),
-                redis_client.clone(),
+                redis_reader_client.clone(),
             ));
 
             let app = feature_flags::router::router(
-                redis_client,
+                redis_reader_client,
+                redis_writer_client,
                 reader,
                 writer,
                 cohort_cache,


Code review comments:
diff_hunk: @@ -337,7 +337,7 @@ mod tests {
             .extract_token()
             .expect("failed to extract token");
 
-        let flag_service = FlagService::new(redis_client.clone(), pg_client.clone());
+        let flag_service = FlagService::new(redis_client.clone(), redis_client.clone(), pg_client.clone());
path: rust/feature-flags/src/flags/flag_request.rs
text: @user1:
logic: Using same Redis client instance for both reader and writer in tests. Should test with separate mock clients to verify split functionality.

@author:
done

diff_hunk: @@ -79,15 +82,15 @@ impl FlagService {
     /// If the team is not found in the cache, it will be fetched from the database and stored in the cache.
     /// Returns the team if found, otherwise an error.
     pub async fn get_team_from_cache_or_pg(&self, token: &str) -> Result<Team, FlagError> {
-        let (team_result, cache_hit) = match Team::from_redis(self.redis_client.clone(), token)
+        let (team_result, cache_hit) = match Team::from_redis(self.redis_reader.clone(), token)
             .await
         {
             Ok(team) => (Ok(team), true),
             Err(_) => match Team::from_pg(self.pg_client.clone(), token).await {
                 Ok(team) => {
                     inc(DB_TEAM_READS_COUNTER, &[], 1);
                     // If we have the team in postgres, but not redis, update redis so we're faster next time
-                    if (Team::update_redis_cache(self.redis_client.clone(), &team).await).is_err() {
+                    if (Team::update_redis_cache(self.redis_writer.clone(), &team).await).is_err() {
path: rust/feature-flags/src/flags/flag_service.rs
text: @user1:
style: Remove unnecessary parentheses around the await expression

```suggestion
                    if Team::update_redis_cache(self.redis_writer.clone(), &team).await.is_err() {
```