idempotent

README

Overview

The provided code implements an idempotent mechanism using Redis to ensure that requests are executed only once, even if they are received multiple times. This is achieved by storing a request identifier and the corresponding response in Redis, and only executing the request again if the stored request identifier does not match the current request identifier.

Key Components

1. Idempotent Struct: The Idempotent struct encapsulates the Redis client and configuration parameters for idempotent requests.

2. Do Method:** The Do method takes a key, a request function, and the request parameter and executes the request idempotently. It first checks if the request has already been executed by retrieving the response from Redis based on the key and the request identifier. If the response exists and the request identifier matches, it returns the stored response. Otherwise, it acquires a lock on the key to prevent duplicate executions, executes the provided request function, and stores the response in Redis.

3. replace Method:** The replace method updates the value of a key in Redis with a new value, but only if the existing value matches the provided old value. This is used to ensure that idempotent requests are not overwritten by subsequent requests.

4. load Method:** The load method retrieves the value of a key from Redis and unmarshals it into a data struct. This is used to retrieve the stored response for an idempotent request.

5. hashRequest Method:** The hashRequest method generates a hash of the provided request parameter. This hash is used as the request identifier to identify idempotent requests.

6. lock Method:** The lock method attempts to acquire a lock on a key using a unique lock value. This prevents multiple concurrent executions of idempotent requests for the same key.

7. unlock Method:** The unlock method releases the lock on a key using the same lock value that was used to acquire the lock. This ensures that other requests can acquire the lock and execute idempotently.

8. hash Method:** The hash method generates a SHA-256 hash of the provided data. This is used to generate the request identifier and the lock value.

9. isUUID Method:** The isUUID method checks if the provided byte slice represents a valid UUID. This is used to validate the lock value.

10. formatMs Method:** The formatMs method converts a time duration to milliseconds. This is used to specify the lock TTL (time-to-live) and the keep TTL for idempotent requests.

11. parseScriptResult Method:** The parseScriptResult method interprets the result of a Redis script and returns an error if it indicates a failure. This is used to handle errors from the Redis scripts used for locking and updating values.

Overall Evaluation

The provided code implements an idempotent mechanism in a clear and well-structured manner. The use of Redis scripts for locking and updating values ensures atomicity and consistency of idempotent requests. The code is also well-documented with comments explaining the purpose of each function and variable.

Documentation

Overview

Package idempotent provides a mechanism for executing requests idempotently using Redis.

Index

• Variables
• type Idempotent
  • func New[K comparable, V any](client *redis.Client, opt *Option) *Idempotent[K, V]
  • func (i *Idempotent[K, V]) Do(ctx context.Context, key string, ...) (res V, err error)
• type Option

Examples

• Idempotent

Variables

var (
	// ErrRequestInFlight indicates that a request is already in flight for the
	// specified key.
	ErrRequestInFlight = errors.New("idempotent: request in flight")

	// ErrRequestMismatch indicates that the request does not match the stored
	// request for the specified key.
	ErrRequestMismatch = errors.New("idempotent: request mismatch")

	// ErrKeyReleased indicates that the key has been
	// released.
	ErrKeyReleased = errors.New("idempotent: key released")
)

Types

type Idempotent

type Idempotent[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/alextanhongpin/core/dsync/idempotent"
	"github.com/alextanhongpin/core/storage/redis/redistest"
	redis "github.com/redis/go-redis/v9"
)

type Request struct {
	Age  int
	Name string
}

type Response struct {
	UserID int64
}

func main() {
	ctx := context.Background()
	stop := redistest.Init()
	defer stop()

	client := redis.NewClient(&redis.Options{
		Addr: redistest.Addr(),
	})
	client.FlushAll(ctx)

	defer client.Close()

	idem := idempotent.New[Request, *Response](client, nil)

	// Create a request object with the required fields.
	req := Request{
		Age:  10,
		Name: "john",
	}

	// Create a wait group to manage concurrent requests.
	var wg sync.WaitGroup
	wg.Add(3)

	// Start the first concurrent request.
	go func() {
		defer wg.Done()

		// Define the handler function that simulates the actual task
		h := func(ctx context.Context, req Request) (*Response, error) {
			fmt.Printf("Executing get user #1: %+v\n", req)
			time.Sleep(40 * time.Millisecond)
			return &Response{UserID: 10}, nil
		}

		// Execute the idempotent operation and handle the response
		v, err := idem.Do(ctx, "get-user", h, req)
		if err != nil {
			panic(err)
		}

		fmt.Printf("Success #1: %+v\n", v)
	}()

	// Start the second concurrent request.
	go func() {
		defer wg.Done()

		// Introduce a delay to simulate a second concurrent request.
		time.Sleep(25 * time.Millisecond)

		// Define the handler function that simulates the actual task.
		h := func(ctx context.Context, req Request) (*Response, error) {
			fmt.Printf("Executing get user #2: %+v\n", req)
			return &Response{UserID: 10}, nil
		}

		// Execute the idempotent operation and handle the response.
		_, err := idem.Do(ctx, "get-user", h, req)
		if err == nil {
			fmt.Println(err)
			panic("want error, got nil")
		}

		// Check if the error is the expected ErrRequestInFlight.
		fmt.Println("Failed #2:", err)
		fmt.Println(errors.Is(err, idempotent.ErrRequestInFlight))
	}()

	// Start the third concurrent request.
	go func() {
		defer wg.Done()

		// Introduce a delay to simulate a third concurrent request.
		// This request happens after the first request completes.
		time.Sleep(60 * time.Millisecond)

		// Define the handler function that simulates the actual task.
		h := func(ctx context.Context, req Request) (*Response, error) {
			fmt.Printf("Executing get user #3: %+v\n", req)
			return &Response{UserID: 10}, nil
		}

		// Execute the idempotent operation and handle the response.
		v, err := idem.Do(ctx, "get-user", h, req)
		if err != nil {
			panic(err)
		}

		fmt.Printf("Success #3: %+v\n", v)
	}()

	wg.Wait()
}

Output:

Executing get user #1: {Age:10 Name:john}
Failed #2: idempotent: request in flight
true
Success #1: &{UserID:10}
Success #3: &{UserID:10}

func New

func New[K comparable, V any](client *redis.Client, opt *Option) *Idempotent[K, V]

New creates a new Idempotent instance with the specified Redis client, lock TTL, and keep TTL.

func (*Idempotent[K, V]) Do

func (i *Idempotent[K, V]) Do(ctx context.Context, key string, fn func(ctx context.Context, req K) (V, error), req K) (res V, err error)

Do executes the provided function idempotently, using the specified key and request.

type Option

type Option struct {
	LockTTL time.Duration
	KeepTTL time.Duration
}