diff --git a/internal/app/token/cache.go b/internal/app/token/cache.go index 296c4681a..c549c95d4 100644 --- a/internal/app/token/cache.go +++ b/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 diff --git a/internal/app/token/verify.go b/internal/app/token/verify.go index 4206b5ad7..d8a46775b 100644 --- a/internal/app/token/verify.go +++ b/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 } diff --git a/internal/app/token/verify_test.go b/internal/app/token/verify_test.go new file mode 100644 index 000000000..32736b9f5 --- /dev/null +++ b/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) + } +} diff --git a/tracking/src/kafka.rs b/tracking/src/kafka.rs index b43b0be34..4d4d0eec9 100644 --- a/tracking/src/kafka.rs +++ b/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, @@ -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()); + } +}