Caching: layers, patterns and what breaks

// Get returns a product, from Redis when it is there and from Postgres otherwise.
func (c *Cache) Get(ctx context.Context, id int64) (Product, error) {
  if p, hit, err := c.Peek(ctx, id); hit || err != nil {
    return p, err // a cached "not found" returns ErrNotFound here
  }
  switch c.strategy {
  case CacheAside:
    return c.fill(ctx, id)
  case Singleflight:
    p, err, _ := c.flights.Do(Key(id), func() (Product, error) { return c.fill(ctx, id) })
    return p, err
  case Lock:
    return c.fillUnderLock(ctx, id)
  }
  return Product{}, fmt.Errorf("unknown strategy %q", c.strategy)
}

// fill reads Postgres and stores the result, or a tombstone when the row does not exist.
func (c *Cache) fill(ctx context.Context, id int64) (Product, error) {
  p, err := c.store.Load(ctx, id)
  if errors.Is(err, ErrNotFound) {
    if err := c.rdb.Set(ctx, Key(id), tombstone, c.NegativeTTL).Err(); err != nil {
      return Product{}, fmt.Errorf("cache tombstone %d: %w", id, err)
    }
    return Product{}, ErrNotFound
  }
  if err != nil {
    return Product{}, err
  }
  return p, c.Put(ctx, p, c.TTL)
}

// UpdatePrice is the write path: change the row, commit, and only then delete the cache key and
// its lease. The next reader fills the cache from the committed row.
func (c *Cache) UpdatePrice(ctx context.Context, db *pgxpool.Pool, id, price int64) (version int64, err error) {
  err = db.QueryRow(ctx, sqls["update_price"], id, price).Scan(&version) // commits on its own
  if errors.Is(err, pgx.ErrNoRows) {
    return 0, ErrNotFound
  }
  if err != nil {
    return 0, fmt.Errorf("update product %d: %w", id, err)
  }
  return version, c.Invalidate(ctx, id)
}

// Invalidate deletes the cached value and any open lease, in one command.
func (c *Cache) Invalidate(ctx context.Context, id int64) error {
  if err := c.rdb.Del(ctx, Key(id), leaseKey(id)).Err(); err != nil {
    return fmt.Errorf("invalidate %d: %w", id, err)
  }
  return nil
}

// Do runs fn for key, or waits for the call already running for key. shared is true when the
// result came from another caller's call.
func (g *Group[V]) Do(key string, fn func() (V, error)) (v V, err error, shared bool) {
  g.mu.Lock()
  if c, ok := g.calls[key]; ok {
    g.mu.Unlock()
    <-c.done // join the call in flight
    return c.val, c.err, true
  }
  c := &call[V]{done: make(chan struct{})}
  if g.calls == nil {
    g.calls = map[string]*call[V]{}
  }
  g.calls[key] = c
  g.mu.Unlock()

  c.val, c.err = fn() // only this caller queries
  g.mu.Lock()
  delete(g.calls, key)
  g.mu.Unlock()
  close(c.done)
  return c.val, c.err, false
}

func (c *Cache) fillUnderLock(ctx context.Context, id int64) (Product, error) {
  token := rand.Text()
  for {
    won, err := c.rdb.SetNX(ctx, lockKey(id), token, c.LockTTL).Result() // SET NX PX
    if err != nil {
      return Product{}, fmt.Errorf("take fill lock %d: %w", id, err)
    }
    if won {
      return c.fillAndUnlock(ctx, id, token)
    }
    // Another reader is filling. Wait for the value, or for the lock to vanish.
    for {
      if err := sleep(ctx, c.PollEvery); err != nil {
        return Product{}, err
      }
      if p, hit, err := c.Peek(ctx, id); hit || err != nil {
        return p, err
      }
      n, err := c.rdb.Exists(ctx, lockKey(id)).Result()
      if err != nil {
        return Product{}, fmt.Errorf("check fill lock %d: %w", id, err)
      }
      if n == 0 {
        break // the holder died without filling: try to take the lock
      }
    }
  }
}

func (c *Cache) fillAndUnlock(ctx context.Context, id int64, token string) (p Product, err error) {
  defer func() {
    if uerr := unlock.Run(ctx, c.rdb, []string{lockKey(id)}, token).Err(); uerr != nil {
      err = errors.Join(err, fmt.Errorf("release fill lock %d: %w", id, uerr))
    }
  }()
  // Check again: the previous holder may have filled the key just before we took the lock.
  if p, hit, err := c.Peek(ctx, id); hit || err != nil {
    return p, err
  }
  return c.fill(ctx, id)
}

-- Release a lock only if this caller still owns it.
-- KEYS[1] the lock key, ARGV[1] the caller's token. Returns 1 if released.
if redis.call('GET', KEYS[1]) == ARGV[1] then
  return redis.call('DEL', KEYS[1])
end
return 0

// JitterTTL returns base plus a random extra time in [0, spread), rounded down to a multiple of
// step. Keys written together then expire at different times. Use step = time.Millisecond in
// production; the stampede experiment uses a coarser step so each expiry lands well inside one
// 100 ms bucket.
func JitterTTL(base, spread, step time.Duration, r *rand.Rand) time.Duration {
  if spread <= 0 || step <= 0 {
    return base
  }
  return base + time.Duration(r.Int64N(int64(spread/step)))*step
}

-- Read a cached value, or take the lease to fill it.
-- KEYS[1] the value key, KEYS[2] its lease key
-- ARGV[1] the caller's lease token, ARGV[2] lease time to live in milliseconds
-- Returns {1, value} on a hit, {2} if the caller now holds the lease, {3} if another caller does.
local v = redis.call('GET', KEYS[1])
if v then
  return {1, v}
end
if redis.call('SET', KEYS[2], ARGV[1], 'NX', 'PX', ARGV[2]) then
  return {2}
end
return {3}

-- Fill the cache only if the caller's lease is still valid.
-- A write deletes the lease, so a fill that read the database before the write is refused.
-- KEYS[1] the value key, KEYS[2] its lease key
-- ARGV[1] lease token, ARGV[2] value, ARGV[3] time to live in milliseconds
-- Returns 1 if the value was stored, 0 if the lease was gone.
if redis.call('GET', KEYS[2]) ~= ARGV[1] then
  return 0
end
redis.call('SET', KEYS[1], ARGV[2], 'PX', ARGV[3])
redis.call('DEL', KEYS[2])
return 1

// RefreshEarly is probabilistic early expiration (the XFetch rule). A reader that hits a key
// refreshes it before it expires with a probability that rises as expiry nears. delta is how long
// the refresh takes; beta above 1 refreshes earlier. Few readers refresh, and most refresh just
// before expiry, so no crowd misses at once.
func RefreshEarly(now, expiry time.Time, delta time.Duration, beta float64, r *rand.Rand) bool {
  u := 1 - r.Float64() // in (0, 1], so the log is finite
  gap := time.Duration(float64(delta) * beta * -math.Log(u))
  return !now.Add(gap).Before(expiry)
}