fix(pier): an ended session is never written back by a request still in flight

- persisting a session's protocol state after a request now rewrites an
  existing record only (Redis SET XX, the memory store likewise) and closes
  the local copy when the record is gone
- before, a request that finished after DELETE, a revoked or deleted subject
  event or the idle expiry put the record back, and the next request
  rehydrated the ended session
- found by plan 19.7-05's TestFanOut (revocation and deletion cases, about
  one run in a hundred)
This commit is contained in:
Jakub Zych
2026-10-23 13:55:04 +02:00
parent c29d2289d0
commit 1111828e8a
4 changed files with 69 additions and 1 deletions

View File

@@ -338,10 +338,24 @@ func (sr *sessionRouter) persist(ctx context.Context, ls *localSession) {
if err != nil {
return
}
if err := sr.p.store.Put(ctx, ls.id, Record{Binding: ls.binding, State: raw, Created: ls.created}, sr.p.cfg.IdleTTL); err != nil {
rec := Record{Binding: ls.binding, State: raw, Created: ls.created}
stored := true
// The record is only rewritten, never created here: a session that ended
// while this request ran (DELETE, revocation, deletion, expiry) stays
// ended instead of being written back by its last request.
if u, ok := sr.p.store.(updater); ok {
stored, err = u.Update(ctx, ls.id, rec, sr.p.cfg.IdleTTL)
} else {
err = sr.p.store.Put(ctx, ls.id, rec, sr.p.cfg.IdleTTL)
}
if err != nil {
logger().Warn("pier: session state not stored", "session", sessionPrefix(ls.id), "error", err)
return
}
if !stored {
sr.closeLocal(ls.id)
return
}
if complete {
ls.persisted.Store(true)
}

View File

@@ -44,3 +44,11 @@ type Store interface {
Publish(ctx context.Context, ev Event) error
Subscribe(ctx context.Context, fn func(Event)) error
}
// updater is implemented by the stores pier ships: Update rewrites an
// existing record and never creates one (Redis SET XX), answering false when
// the record is gone, so a request that finishes after its session ended
// (DELETE, revocation, deletion, expiry) cannot write the record back.
type updater interface {
Update(ctx context.Context, id string, rec Record, ttl time.Duration) (bool, error)
}

View File

@@ -52,6 +52,19 @@ func (m *memoryStore) Get(_ context.Context, id string) (Record, bool, error) {
return r.rec, true, nil
}
// Update rewrites an existing record; a missing or expired one stays gone.
func (m *memoryStore) Update(_ context.Context, id string, rec Record, ttl time.Duration) (bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
r, ok := m.records[id]
if !ok || !r.expires.After(m.now()) {
delete(m.records, id)
return false, nil
}
m.records[id] = memoryRecord{rec: rec, expires: m.now().Add(ttl)}
return true, nil
}
func (m *memoryStore) Touch(_ context.Context, id string, ttl time.Duration) error {
m.mu.Lock()
defer m.mu.Unlock()

View File

@@ -51,6 +51,27 @@ func (s *redisStore) Put(ctx context.Context, id string, rec Record, ttl time.Du
return err
}
// Update rewrites an existing session record (SET XX) and refreshes the
// binding set; a record that is gone stays gone.
func (s *redisStore) Update(ctx context.Context, id string, rec Record, ttl time.Duration) (bool, error) {
raw, err := json.Marshal(rec)
if err != nil {
return false, err
}
err = s.client.SetArgs(ctx, s.sessionKey(id), raw, redis.SetArgs{Mode: "XX", TTL: ttl}).Err()
if errors.Is(err, redis.Nil) {
return false, nil
}
if err != nil {
return false, err
}
pipe := s.client.TxPipeline()
pipe.SAdd(ctx, s.bindingKey(rec.Binding), id)
pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl)
_, err = pipe.Exec(ctx)
return true, err
}
func (s *redisStore) Get(ctx context.Context, id string) (Record, bool, error) {
raw, err := s.client.Get(ctx, s.sessionKey(id)).Bytes()
if errors.Is(err, redis.Nil) {
@@ -198,6 +219,18 @@ func (l *lazyStore) Get(ctx context.Context, id string) (Record, bool, error) {
return s.Get(ctx, id)
}
// Update forwards to the opened store's Update (Put for a store without one).
func (l *lazyStore) Update(ctx context.Context, id string, rec Record, ttl time.Duration) (bool, error) {
s, err := l.get(ctx)
if err != nil {
return false, err
}
if u, ok := s.(updater); ok {
return u.Update(ctx, id, rec, ttl)
}
return true, s.Put(ctx, id, rec, ttl)
}
func (l *lazyStore) Touch(ctx context.Context, id string, ttl time.Duration) error {
s, err := l.get(ctx)
if err != nil {