Distributed cache

// detect closes one window for an app server: every key whose Space-Saving count is above
// HotShare of the reads it saw becomes hot for the next window. The sketch holds Counters
// keys, so the memory stays fixed however many keys pass through.
func detect(a *app, cfg Config) {
  hot := map[string]bool{}
  for _, it := range a.ss.Top(cfg.Counters) {
    if float64(it.Count) > cfg.HotShare*float64(a.n) {
      hot[it.Key] = true
    }
  }
  a.hot, a.n = hot, 0
  a.ss = sketches.NewSpaceSaving(cfg.Counters)
}

// cacheKey picks where a read goes. A key the app server marked hot has Copies copies,
// key#0 to key#(Copies-1); each read takes one at random, so the copies hash to different
// nodes and share the load. Every other key has one home on the ring.
func cacheKey(a *app, cfg Config, key string, r *rand.Rand) string {
  if cfg.Replicate && a.hot[key] {
    return fmt.Sprintf("%s#%d", key, r.IntN(cfg.Copies))
  }
  return key
}

res.Reads++
reads[k]++
bReads++
a.ss.Add(key)
a.n++
if cfg.L1 {
  if v, ok := a.l1.Get(key); ok { // 1. this app server's memory
    res.L1Hits++
    bHits++
    if v < version[k] {
      res.StaleReads++
    }
    continue
  }
}
ck := cacheKey(a, cfg, key, r)
n := nodes[ring.Owner(ck)] // 2. the cache node that owns the key
res.NodeLoad[n]++
bNode[n]++
v, ok := caches[n].Get(ck)
if ok {
  res.NodeHits++
  bHits++
  if v < version[k] {
    res.StaleReads++
  }
} else { // 3. the database, then fill the cache
  res.DBReads++
  bl.DBReads++
  v = version[k]
  caches[n].Put(ck, v, time.Hour)
}
if cfg.L1 {
  a.l1.Put(key, v, cfg.L1TTL)
}

// Add counts one occurrence of key. A key in the table gets +1. A new key takes a free counter,
// or replaces the key with the smallest count and inherits that count + 1. The inherited part is
// its error bound.
func (s *SpaceSaving) Add(key string) {
  s.n++
  if i, ok := s.h.at[key]; ok {
    s.h.items[i].Count++
    heap.Fix(&s.h, i)
    return
  }
  if s.h.Len() < s.cap {
    heap.Push(&s.h, &Item{Key: key, Count: 1})
    return
  }
  low := s.h.items[0]
  delete(s.h.at, low.Key)
  s.h.at[key] = 0
  *low = Item{Key: key, Count: low.Count + 1, Err: low.Count}
  heap.Fix(&s.h, 0)
}

// write is cache-aside with delete on write: after the database commits, delete the key and
// every copy it may have on the ring. The next read fills from the database.
func write(caches []*caching.LRU[string, int64], ring *sharding.Ring, nodes map[string]int, cfg Config, key string) {
  drop(caches[nodes[ring.Owner(key)]], key)
  if cfg.Replicate {
    for c := range cfg.Copies {
      ck := fmt.Sprintf("%s#%d", key, c)
      drop(caches[nodes[ring.Owner(ck)]], ck)
    }
  }
}

// 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
}

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)
}