upstream-seaweedfs/weed/s3api/s3lifecycle/reader/cursor.go

package reader

import (
	"sync"

	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
)

// Cursor tracks the meta-log position for each ActionKey within one shard.
// Position = tsNs of the last event the shard's reader+downstream considers
// "fully resolved" for this ActionKey. Subscription resumes at the minimum
// position across all keys.
//
// A frozen key is pinned at its current position (BLOCKED outcome): Advance
// is a no-op until Unfreeze. The meta-log retention bounds how long a freeze
// can last before the entry expires from the log; operator action via the
// blocker-resolve flow lifts the freeze.
type Cursor struct {
	mu     sync.RWMutex
	state  map[s3lifecycle.ActionKey]int64
	frozen map[s3lifecycle.ActionKey]struct{}
}

func NewCursor() *Cursor {
	return &Cursor{
		state:  map[s3lifecycle.ActionKey]int64{},
		frozen: map[s3lifecycle.ActionKey]struct{}{},
	}
}

// MinTsNs returns the smallest position across all keys; zero if empty.
// Used as the resume point for the meta-log subscription.
func (c *Cursor) MinTsNs() int64 {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if len(c.state) == 0 {
		return 0
	}
	var min int64 = -1
	for _, v := range c.state {
		if min < 0 || v < min {
			min = v
		}
	}
	if min < 0 {
		return 0
	}
	return min
}

// Get returns the current position for key, or 0 if unset.
func (c *Cursor) Get(key s3lifecycle.ActionKey) int64 {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.state[key]
}

// Advance moves key's position forward to tsNs. No-op if key is frozen, if
// tsNs <= current, or if tsNs <= 0.
func (c *Cursor) Advance(key s3lifecycle.ActionKey, tsNs int64) {
	if tsNs <= 0 {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, isFrozen := c.frozen[key]; isFrozen {
		return
	}
	if cur := c.state[key]; tsNs > cur {
		c.state[key] = tsNs
	}
}

// Freeze pins key at its current position. If key has no recorded position
// yet, it is set to tsNs.
func (c *Cursor) Freeze(key s3lifecycle.ActionKey, tsNs int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, exists := c.state[key]; !exists {
		c.state[key] = tsNs
	}
	c.frozen[key] = struct{}{}
}

// Unfreeze releases a freeze. Subsequent Advance calls take effect.
func (c *Cursor) Unfreeze(key s3lifecycle.ActionKey) {
	c.mu.Lock()
	defer c.mu.Unlock()
	delete(c.frozen, key)
}

// IsFrozen reports whether key is currently pinned.
func (c *Cursor) IsFrozen(key s3lifecycle.ActionKey) bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	_, ok := c.frozen[key]
	return ok
}

// Snapshot copies the cursor map for persistence. Frozen keys are not
// distinguished in the snapshot; callers persist freezes via the separate
// blocker-record store.
func (c *Cursor) Snapshot() map[s3lifecycle.ActionKey]int64 {
	c.mu.RLock()
	defer c.mu.RUnlock()
	out := make(map[s3lifecycle.ActionKey]int64, len(c.state))
	for k, v := range c.state {
		out[k] = v
	}
	return out
}

// Restore replaces the cursor map. Freezes are not restored here; the
// caller re-applies them from blocker records on startup.
func (c *Cursor) Restore(state map[s3lifecycle.ActionKey]int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.state = make(map[s3lifecycle.ActionKey]int64, len(state))
	for k, v := range state {
		c.state[k] = v
	}
	c.frozen = map[s3lifecycle.ActionKey]struct{}{}
}