redis_transaction_with_watch_and_retry.go

package main

import (
	"context"
	"errors"
	"fmt"
	"github.com/redis/go-redis/v9"
	"sync"
	"time"
)

func MakeNewPage(ctx context.Context, rdb *redis.Client, slug string, viewLimit int) error {
	if err := rdb.Set(ctx, fmt.Sprintf("page:%s:views", slug), 0, 0).Err(); err != nil {
		return fmt.Errorf("error while saving page default view: %v", err)
	}
	if err := rdb.Set(ctx, fmt.Sprintf("page:%s:viewLimit", slug), viewLimit, 0).Err(); err != nil {
		fmt.Errorf("error while setting page view limit: %v", err)
	}
	return nil
}

func CheckIfCanVisitPageWithoutTransaction(ctx context.Context, rdb *redis.Client, slug string) (bool, error) {
	viewLimitKey := fmt.Sprintf("page:%s:viewLimit", slug)
	viewsKey := fmt.Sprintf("page:%s:views", slug)

	canView := false
	return canView, rdb.Watch(ctx, func(tx *redis.Tx) error {
		// using tx instead of rdb ensures if those values has changed, the transaction will fail
		limit, err := tx.Get(ctx, viewLimitKey).Int()
		if err != nil {
			return fmt.Errorf("error while getting page view limit: %v", err)
		}

		currentViews, err := tx.Get(ctx, viewsKey).Int()
		if err != nil {
			return fmt.Errorf("error while getting page's current views: %v", err)
		}

		<-time.After(time.Second)

		// the page has reached its view limit
		if currentViews >= limit {
			return nil
		}

		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			// adding the new view
			if err := pipe.Set(ctx, viewsKey, currentViews+1, 0).Err(); err != nil {
				// if an error happens the view has not been added, and we don't show the user the page
				return fmt.Errorf("error while saving page default view: %v", err)
			}
			return nil
		})

		if err != nil {
			return err
		}
		canView = true
		return nil

	}, viewsKey)

}

var (
	ErrMaxRetriesReached = errors.New("max retries reached")
)

func TransactionWithRetry(callback func() error, maxRetries int) error {
	retries := 0
	for {
		err := callback()
		// the transaction executed successfully
		if err == nil {
			return nil
		}

		if errors.Is(err, redis.TxFailedErr) {
			retries++
			fmt.Println("> retry happened.")
			if retries > maxRetries {
				return ErrMaxRetriesReached
			}
			continue
		}

		// something unexpected happened
		return err

	}
}

func main() {
	rdb := redis.NewClient(&redis.Options{})

	if err := MakeNewPage(context.Background(), rdb, "test-1", 10); err != nil {
		panic(err)
	}

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		err := TransactionWithRetry(func() error {
			_, err := CheckIfCanVisitPageWithoutTransaction(context.Background(), rdb, "test-1")
			return err
		}, 2)
		if err != nil {
			panic(err)
		}
		wg.Done()
	}()
	go func() {
		<-time.After(time.Millisecond * 500)
		err := TransactionWithRetry(func() error {
			_, err := CheckIfCanVisitPageWithoutTransaction(context.Background(), rdb, "test-1")
			return err
		}, 2)
		if err != nil {
			panic(err)
		}
		wg.Done()
	}()
	wg.Wait()

}