transaction.go

// Copyright 2022 Molecula Corp. (DBA FeatureBase).
// SPDX-License-Identifier: Apache-2.0
package pilosa

import (
	"context"
	"encoding/json"
	"regexp"
	"sync"
	"time"

	"github.com/featurebasedb/featurebase/v3/logger"
	"github.com/pkg/errors"
)

var txIDRegexp = regexp.MustCompile("^[A-Za-z0-9_-]*$")

// Transaction contains information related to a block of work that
// needs to be tracked and spans multiple API calls.
type Transaction struct {
	// ID is an arbitrary string identifier. All transactions must have a unique ID.
	ID string `json:"id"`

	// Active notes whether an exclusive transaction is active, or
	// still pending (if other active transactions exist). All
	// non-exclusive transactions are always active.
	Active bool `json:"active"`

	// Exclusive is set to true for transactions which can only become active when no other
	// transactions exist.
	Exclusive bool `json:"exclusive"`

	// Timeout is the minimum idle time for which this transaction should continue to exist.
	Timeout time.Duration `json:"timeout"`

	// CreatedAt is the timestamp at which the transaction was created. This supports
	// the case of listing transactions in a useful order.
	CreatedAt time.Time `json:"createdAt"`

	// Deadline is calculated from Timeout. TODO reset deadline each time there is activity
	// on the transaction. (we can't do this until there is some method of associating a
	// request/call with a transaction)
	Deadline time.Time `json:"deadline"`

	// Stats track statistics for the transaction. Not yet used.
	Stats TransactionStats `json:"stats"`
}

type TransactionStats struct{}

// TransactionManager enforces the rules for transactions on a single
// node. It is goroutine-safe. It should be created by a call to
// NewTransactionManager where it takes a TransactionStore. If logging
// is desired, Log should be set before an instance of
// TransactionManager is used.
type TransactionManager struct {
	mu sync.RWMutex

	Log logger.Logger

	store TransactionStore

	checkingDeadlines bool
}

// NewTransactionManager creates a new TransactionManager with the
// given store, and starts a deadline-checker in a goroutine.
func NewTransactionManager(store TransactionStore) *TransactionManager {
	tm := &TransactionManager{
		Log:               logger.NopLogger,
		store:             store,
		checkingDeadlines: true,
	}
	// start deadline checker in case we've just started up, but there is already state in the store.
	go tm.deadlineChecker()
	return tm
}

// Start starts a new transaction with the given parameters. If an
// exclusive transaction is pending or in progress,
// ErrTransactionExclusive is returned. If a transaction with the same
// id already exists, that transaction is returned along with
// ErrTransactionExists. If there is no error, the created transaction
// is returned—this is primarily so that the caller can discover if an
// exclusive transaction has been made immediately active or if they
// need to poll.
func (tm *TransactionManager) Start(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) {
	tm.mu.Lock()
	defer tm.mu.Unlock()

	if !txIDRegexp.Match([]byte(id)) {
		return nil, errors.New("invalid transaction ID, must match [A-Za-z0-9_-]")
	}

	trnsMap, err := tm.store.List()
	if err != nil {
		return nil, errors.Wrap(err, "listing transactions in Start")
	}

	// check for an exclusive transaction
	for _, trns := range trnsMap {
		if trns.Exclusive {
			// if someone wants a transaction, and we're not able to
			// give it to them, we want to be checking deadlines.
			// TODO: it would be nice if we could identify whether
			// this trns has expired, and if so, automatically remove
			// it and continue without returning ErrTransactionExclusive
			// on this iteration of the loop. One way we could do that is
			// to call tm.checkDeadlines() here (note that we'd have to
			// have an unprotectedCheckDeadlines()), and then after that
			// check if trns still exists in tm.store. If not, continue.
			tm.startDeadlineChecker()
			return trns, ErrTransactionExclusive
		}
	}
	if trns, ok := trnsMap[id]; ok {
		return trns, ErrTransactionExists
	}

	// set new transaction to active if it is not exclusive or if
	// there are no other transactions.
	active := !exclusive || (len(trnsMap) == 0)

	// set deadline according to timeout
	createdAt := time.Now()
	deadline := createdAt.Add(timeout)
	trns := &Transaction{
		ID:        id,
		Active:    active,
		Exclusive: exclusive,
		Timeout:   timeout,
		CreatedAt: createdAt,
		Deadline:  deadline,
	}
	if err = tm.store.Put(trns); err != nil {
		return nil, errors.Wrap(err, "adding to store")
	}

	// we won't check deadlines unless there's actually a
	// transaction pending
	if exclusive && !active {
		tm.startDeadlineChecker()
	}

	return trns, nil
}

// Finish completes and removes a transaction, returning the completed
// transaction (so that the caller can e.g. view the Stats)
func (tm *TransactionManager) Finish(ctx context.Context, id string) (*Transaction, error) {
	tm.mu.Lock()
	defer tm.mu.Unlock()
	return tm.finish(id)
}

// finish is the unprotected implementation of Finish
func (tm *TransactionManager) finish(id string) (*Transaction, error) {
	trns, err := tm.store.Remove(id)
	if err != nil {
		return nil, err
	}

	// After removing, check to see if we need to activate an exclusive transaction
	trnsMap, err := tm.store.List()
	if err != nil {
		tm.log().Errorf("error listing transactions in Finish: %v", err)
		return trns, nil
	}

	if len(trnsMap) == 1 {
		for _, etrans := range trnsMap {
			if etrans.Exclusive {
				if etrans.Active { // sanity check
					panic("we just removed a transaction, and the sole remaining exclusive transaction was already active")
				}
				etrans.Active = true
				etrans.Deadline = time.Now().Add(etrans.Timeout)
				if err := tm.store.Put(etrans); err != nil {
					tm.log().Errorf("activating exclusive transaction after finishing last transaction: %v", err)
					return trns, nil
				}
			}
		}
	}
	return trns, nil
}

// Get retrieves the transaction with the given ID. Returns ErrTransactionNotFound
// if there isn't one.
func (tm *TransactionManager) Get(ctx context.Context, id string) (*Transaction, error) {
	tm.mu.RLock()
	defer tm.mu.RUnlock()

	return tm.store.Get(id)
}

// List returns map of all transactions by their ID. It is a copy and
// so may be retained and modified by the caller.
func (tm *TransactionManager) List(ctx context.Context) (map[string]*Transaction, error) {
	tm.mu.RLock()
	defer tm.mu.RUnlock()
	return tm.store.List()
}

// ResetDeadline updates the deadline for the transaction with the
// given ID to be equal to the current time plus the transaction's
// timeout.
func (tm *TransactionManager) ResetDeadline(ctx context.Context, id string) (*Transaction, error) {
	tm.mu.Lock()
	defer tm.mu.Unlock()
	trns, err := tm.store.Get(id)
	if err != nil {
		return nil, errors.Wrap(err, "getting transaction")
	}

	trns.Deadline = time.Now().Add(trns.Timeout)

	err = tm.store.Put(trns)
	return trns, errors.Wrap(err, "storing transaction with new timeout")
}

// startDeadlineChecker may only be called while tm.mu is held.
func (tm *TransactionManager) startDeadlineChecker() {
	if !tm.checkingDeadlines {
		tm.checkingDeadlines = true
		go tm.deadlineChecker()
	}
}

// deadlineChecker loops continuously checking for expired
// deadlines. It stops when there are no upcoming deadlines.
func (tm *TransactionManager) deadlineChecker() {
	interval := tm.checkDeadlines()
	for interval != 0 {
		time.Sleep(interval)
		interval = tm.checkDeadlines()
	}
	tm.mu.Lock()
	tm.checkingDeadlines = false
	tm.mu.Unlock()
}

// checkDeadlines finishes transactions which are past their
// deadlines. It returns the duration until the next deadline. If
// there are no exclusive transactions, it does nothing and returns 0
// as a signal to stop checking.
func (tm *TransactionManager) checkDeadlines() time.Duration {
	tm.mu.Lock()
	defer tm.mu.Unlock()

	trnsMap, err := tm.store.List()
	if err != nil {
		tm.log().Errorf("transaction deadline checker couldn't list transactions: %v", err)
		return 0
	}

	hasExclusive := false
	for _, trns := range trnsMap {
		if trns.Exclusive {
			hasExclusive = true
			break
		}
	}
	if !hasExclusive {
		return 0 // no need to expire things if nothing is waiting
	}

	now := time.Now()
	// track the time interval to next deadline
	nextInterval := time.Duration(0)
	for id, trns := range trnsMap {
		if !trns.Active {
			continue
		}
		if !now.Before(trns.Deadline) {
			trnsF, err := tm.finish(id)
			if err != nil {
				tm.log().Errorf("error finishing expired transaction '%s': %+v: %v", id, trnsF, err)
			} else {
				tm.log().Infof("cleared expired transaction: %+v", trnsF)
			}
		} else {
			interval := trns.Deadline.Sub(now)
			if nextInterval == 0 || interval < nextInterval {
				nextInterval = interval
			}
		}
	}
	return nextInterval
}

func (tm *TransactionManager) log() logger.Logger {
	if tm.Log != nil {
		return tm.Log
	}
	return logger.NopLogger
}

// TransactionStore declares the functionality which a store for
// Pilosa transactions must implement.
type TransactionStore interface {
	// Put stores a new transaction or replaces an existing transaction with the given one.
	Put(trns *Transaction) error
	// Get retrieves the transaction at id or returns ErrTransactionNotFound if there isn't one.
	Get(id string) (*Transaction, error)
	// List returns a map of all transactions by ID. The map must be safe to modify by the caller.
	List() (map[string]*Transaction, error)
	// Remove deletes the transaction from the store. It must return ErrTransactionNotFound if there isn't one.
	Remove(id string) (*Transaction, error)
}

type OpenTransactionStoreFunc func(path string) (TransactionStore, error)

func OpenInMemTransactionStore(path string) (TransactionStore, error) {
	return NewInMemTransactionStore(), nil
}

// InMemTransactionStore does not persist transaction data and is only
// useful for testing.
type InMemTransactionStore struct {
	mu   sync.RWMutex
	tmap map[string]*Transaction
}

func NewInMemTransactionStore() *InMemTransactionStore {
	return &InMemTransactionStore{
		tmap: make(map[string]*Transaction),
	}
}

func (s *InMemTransactionStore) Put(trns *Transaction) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.tmap[trns.ID] = trns.Copy()
	return nil
}

func (s *InMemTransactionStore) Get(id string) (*Transaction, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	if trns, ok := s.tmap[id]; ok {
		return trns.Copy(), nil
	}
	return nil, ErrTransactionNotFound
}

func (s *InMemTransactionStore) List() (map[string]*Transaction, error) {
	cp := make(map[string]*Transaction)
	for id, trns := range s.tmap {
		cp[id] = trns.Copy()
	}
	return cp, nil
}

func (s *InMemTransactionStore) Remove(id string) (*Transaction, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if trns, ok := s.tmap[id]; ok {
		delete(s.tmap, id)
		return trns.Copy(), nil
	}
	return nil, ErrTransactionNotFound
}

type Error string

func (e Error) Error() string { return string(e) }

const ErrTransactionNotFound = Error("transaction not found")
const ErrTransactionExclusive = Error("there is an exclusive transaction, try later")
const ErrTransactionExists = Error("transaction with the given id already exists")

func CompareTransactions(t1, t2 *Transaction) error {
	if t1 == nil && t2 == nil {
		return nil
	}
	if t1 == nil || t2 == nil {
		return errors.Errorf("transactions are not equal: %+v %+v", t1, t2)
	}
	if t1.ID != t2.ID {
		return errors.Errorf("transaction IDs not equal: %+v %+v", t1, t2)
	}
	if t1.Active != t2.Active {
		return errors.Errorf("transaction Actives not equal: %+v %+v", t1, t2)
	}
	if t1.Exclusive != t2.Exclusive {
		return errors.Errorf("transaction Exclusives not equal: %+v %+v", t1, t2)
	}
	if t1.Timeout != t2.Timeout {
		return errors.Errorf("transaction Timeouts not equal: %+v %+v", t1, t2)
	}
	// don't care about Deadline or Stats
	return nil
}

func (trns *Transaction) UnmarshalJSON(b []byte) error {
	tmp := &struct {
		ID        string      `json:"id"`
		Active    bool        `json:"active"`
		Exclusive bool        `json:"exclusive"`
		Timeout   interface{} `json:"timeout"`
		Deadline  string      `json:"deadline"`
	}{}
	err := json.Unmarshal(b, tmp)
	if err != nil {
		return err
	}
	trns.ID = tmp.ID
	trns.Active = tmp.Active
	trns.Exclusive = tmp.Exclusive
	switch tm := tmp.Timeout.(type) {
	case string:
		dur, err := time.ParseDuration(tm)
		if err != nil {
			return errors.Wrapf(err, "timeout as string must be a valid duration got: '%s'", tm)
		}
		trns.Timeout = dur
	case float64:
		// interpret as number of seconds
		seconds := int64(tm)
		nsec := (tm - float64(seconds)) * 1e9
		trns.Timeout = time.Duration(seconds*1e9 + int64(nsec))
	case nil:
		break
	default:
		return errors.New("timeout must be float64 or string")
	}

	if tmp.Deadline != "" {
		trns.Deadline, err = time.Parse(time.RFC3339Nano, tmp.Deadline)
	}
	return errors.Wrap(err, "parsing deadline")
}

func (trns *Transaction) MarshalJSON() ([]byte, error) {
	return json.Marshal(&struct {
		ID        string `json:"id"`
		Active    bool   `json:"active"`
		Exclusive bool   `json:"exclusive"`
		Timeout   string `json:"timeout"`
		CreatedAt string `json:"createdAt"`
		Deadline  string `json:"deadline"`
	}{
		ID:        trns.ID,
		Active:    trns.Active,
		Exclusive: trns.Exclusive,
		Timeout:   trns.Timeout.String(),
		CreatedAt: trns.CreatedAt.In(time.UTC).Format(time.RFC3339Nano),
		Deadline:  trns.Deadline.In(time.UTC).Format(time.RFC3339Nano),
	})
}

func (trns *Transaction) Copy() *Transaction {
	return &Transaction{
		ID:        trns.ID,
		Active:    trns.Active,
		Exclusive: trns.Exclusive,
		Timeout:   trns.Timeout,
		CreatedAt: trns.CreatedAt,
		Deadline:  trns.Deadline,
		Stats:     trns.Stats,
	}
}