fix: stop a Redis outage locking everyone out, and let the tracking encoder register its schema (#537)

* feat: treat a session cache that cannot answer as a miss rather than an authentication failure, because the session lives in Postgres and reading it from there is what the caller already falls back to, so a Redis outage costs a database read per request instead of locking every user out of the dashboard with a 500

* feat: carry the tracking event schema with the subject-name strategy so the encoder registers it instead of only looking one up, because a registry with no tracking-events subject had nothing to find and answered every event with 'Could not get id from response', dropping every open and click on the floor

internal/app/token/cache.go

@@ -34,17 +34,24 @@ func (s *tokenService) saveSession(ctx context.Context, session *models.Session,
 func (s *tokenService) getSession(ctx context.Context, sessionID uuid.UUID) (*models.Session, *errx.Error) {
 	data, err := s.cache.Get(ctx, getSessionKey(sessionID)).Bytes()
 	if err != nil {
-		if errors.Is(err, redis.Nil) {
-			return nil, nil
+		if !errors.Is(err, redis.Nil) {
+			// A cache that cannot answer is a miss, not a failed request. The
+			// session lives in Postgres and the caller reads it from there;
+			// treating an unreachable Redis as an auth failure turned a cache
+			// outage into nobody being able to sign in at all.
+			//
+			// Safe on the revocation path too: falling through reads the
+			// authoritative row rather than a cached copy of it.
+			errs.CaptureException(err)
 		}
-		errs.CaptureException(err)
-		return nil, errx.InternalError()
+		return nil, nil
 	}
 
 	var session models.Session
 	if err := json.Unmarshal(data, &session); err != nil {
+		// Unreadable cached bytes are a miss for the same reason.
 		errs.CaptureException(err)
-		return nil, errx.InternalError()
+		return nil, nil
 	}
 
 	return &session, nil

internal/app/token/verify.go

@@ -24,9 +24,11 @@ func (s *tokenService) GetSession(ctx context.Context, sessionID uuid.UUID) (*mo
 		return nil, err
 	}
 
-	if err := s.saveSession(ctx, sess, SessionTTL); err != nil {
-		return nil, err
-	}
+	// Best effort: the session is already resolved, and failing to cache it
+	// only costs the next request a database read. Returning the error here
+	// meant a Redis write that was refused, on quota or anything else, failed
+	// every authenticated request with a 500.
+	_ = s.saveSession(ctx, sess, SessionTTL)
 	return sess, nil
 }
 

internal/app/token/verify_test.go

@@ -0,0 +1,59 @@
+package token
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/infrastructure/cache"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// A cache that cannot answer must not be able to sign everyone out.
+//
+// It could. The session read returned InternalError on any Redis error that was
+// not a plain miss, and the write that followed a database read returned its
+// error too. So when Redis began refusing writes on a quota, every
+// authenticated request failed with a 500 and nobody could log in, while the
+// sessions themselves sat intact in Postgres the whole time.
+//
+// The cache here points at a closed port, which is the real failure rather than
+// a mocked one.
+type sessionRepo struct {
+	repository.TokenRepository
+	session *models.Session
+	calls   int
+}
+
+func (r *sessionRepo) GetSession(_ context.Context, _ uuid.UUID) (*models.Session, *errx.Error) {
+	r.calls++
+	return r.session, nil
+}
+
+func TestASessionSurvivesACacheThatIsDown(t *testing.T) {
+	// Built directly rather than through cache.New, which pings on construction
+	// and would fail before the test could start.
+	down := &cache.Cache{Client: redis.NewClient(&redis.Options{
+		Addr:        "127.0.0.1:1",
+		DialTimeout: 200 * time.Millisecond,
+	})}
+	id := uuid.New()
+	repo := &sessionRepo{session: &models.Session{ID: id, UserID: uuid.New()}}
+	s := &tokenService{cache: down, tokenRepository: repo}
+
+	got, xerr := s.GetSession(context.Background(), id)
+	if xerr != nil {
+		t.Fatalf("a cache outage failed the request: %v", xerr)
+	}
+	if got == nil || got.ID != id {
+		t.Fatalf("session = %+v, want id %s", got, id)
+	}
+	if repo.calls != 1 {
+		t.Fatalf("read the database %d times, want 1", repo.calls)
+	}
+}

tracking/src/kafka.rs

@@ -4,7 +4,9 @@ use rdkafka::producer::{FutureProducer, FutureRecord};
 use rdkafka::util::Timeout;
 use schema_registry_converter::async_impl::avro::AvroEncoder;
 use schema_registry_converter::async_impl::schema_registry::SrSettings;
-use schema_registry_converter::schema_registry_common::SubjectNameStrategy;
+use schema_registry_converter::schema_registry_common::{
+    SchemaType, SubjectNameStrategy, SuppliedSchema,
+};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
@@ -15,7 +17,6 @@ use crate::events::TrackingEvent;
 use crate::observability;
 
 /// Avro schema for tracking events - matches Go events.TrackingEvent
-#[allow(dead_code)]
 pub const TRACKING_EVENT_SCHEMA: &str = r#"
 {
     "type": "record",
@@ -36,6 +37,20 @@ pub const TRACKING_EVENT_SCHEMA: &str = r#"
 }
 "#;
 
+/// The schema as the registry needs it: the document, its type, and the full
+/// record name so a subject registered from here matches one registered by any
+/// other client.
+fn tracking_event_schema() -> SuppliedSchema {
+    SuppliedSchema {
+        name: Some("com.warmbly.tracking.TrackingEvent".to_string()),
+        schema_type: SchemaType::Avro,
+        schema: TRACKING_EVENT_SCHEMA.to_string(),
+        references: vec![],
+        properties: None,
+        tags: None,
+    }
+}
+
 #[derive(Clone)]
 pub struct KafkaProducer {
     producer: Arc<FutureProducer>,
@@ -220,14 +235,82 @@ impl KafkaProducer {
         };
         let encoder = encoder.read().await;
 
-        // Use schema registry encoder to serialize with proper schema ID prefix
+        // The schema travels with the strategy so the encoder registers it on
+        // first use. TopicNameStrategy alone only LOOKS one up, and a subject
+        // that does not exist yet has nothing to find, which a fresh registry
+        // reports as "Could not get id from response" on every single event.
+        // Nothing then reaches the bus and every open and click is dropped.
         let payload = encoder
             .encode(
                 event.to_avro_value(),
-                SubjectNameStrategy::TopicNameStrategy(self.topic.clone(), false),
+                SubjectNameStrategy::TopicNameStrategyWithSchema(
+                    self.topic.clone(),
+                    false,
+                    tracking_event_schema(),
+                ),
             )
             .await?;
 
         Ok(payload)
     }
 }
+
+#[cfg(test)]
+mod registry_tests {
+    use super::*;
+    use crate::events::TrackingEvent;
+
+    /// Against a real Schema Registry, which is the only thing that shows
+    /// whether the encoder can register a subject rather than merely look one
+    /// up. Skipped without one:
+    ///
+    ///   SR_URL=... SR_KEY=... SR_SECRET=... cargo test --features kafka registry
+    ///
+    /// This is the gap that took tracking down during a codec cutover: every
+    /// event failed with "Could not get id from response" because the strategy
+    /// carried no schema, and nothing in CI had ever encoded against a registry.
+    #[tokio::test]
+    async fn encodes_against_a_real_registry() {
+        let Ok(url) = std::env::var("SR_URL") else {
+            return;
+        };
+        let settings = SrSettings::new_builder(url)
+            .set_basic_authorization(
+                &std::env::var("SR_KEY").unwrap_or_default(),
+                Some(&std::env::var("SR_SECRET").unwrap_or_default()),
+            )
+            .build()
+            .expect("settings");
+        let encoder = AvroEncoder::new(settings);
+
+        let event = TrackingEvent {
+            event_type: "EMAIL_OPENED".to_string(),
+            task_id: uuid::Uuid::new_v4().to_string(),
+            original_url: None,
+            link_id: None,
+            timestamp: chrono::Utc::now().to_rfc3339(),
+            user_agent: Some("probe".to_string()),
+            ip_hash: None,
+            client_ip: None,
+            scanner: None,
+            scanner_probable: false,
+        };
+
+        let payload = encoder
+            .encode(
+                event.to_avro_value(),
+                SubjectNameStrategy::TopicNameStrategyWithSchema(
+                    "warmbly-tracking-schema-check".to_string(),
+                    false,
+                    tracking_event_schema(),
+                ),
+            )
+            .await
+            .expect("the encoder must register the subject, not only look it up");
+
+        // Confluent framing: a zero byte then the schema id.
+        assert_eq!(payload[0], 0, "payload is not schema-registry framed");
+        assert!(payload.len() > 5, "payload carries no body");
+        println!("encoded {} bytes against the live registry", payload.len());
+    }
+}