102 lines
3.0 KiB
Go
102 lines
3.0 KiB
Go
|
package idempotency
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"os"
|
||
|
|
"sync"
|
||
|
|
"sync/atomic"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
)
|
||
|
|
|
||
|
|
func TestExecutorReplaysAndRejectsHashConflict(t *testing.T) {
|
||
|
|
pool := integrationPool(t)
|
||
|
|
executor := NewExecutor(pool)
|
||
|
|
key := fmt.Sprintf("test-%d", time.Now().UnixNano())
|
||
|
|
tenantID := "10000000-0000-0000-0000-000000000001"
|
||
|
|
request := Request{
|
||
|
|
TenantID: &tenantID, ActorKey: "integration-user", Scope: "integration.create",
|
||
|
|
Key: key, RequestHash: Hash([]byte(`{"value":1}`)), ExpiresAfter: time.Hour,
|
||
|
|
}
|
||
|
|
var calls atomic.Int32
|
||
|
|
operation := func(context.Context, pgx.Tx) (Result, error) {
|
||
|
|
calls.Add(1)
|
||
|
|
return Result{StatusCode: 201, Body: json.RawMessage(`{"created":true}`)}, nil
|
||
|
|
}
|
||
|
|
first, err := executor.Execute(context.Background(), request, operation)
|
||
|
|
if err != nil || first.Replayed {
|
||
|
|
t.Fatalf("first execute = %#v, %v", first, err)
|
||
|
|
}
|
||
|
|
second, err := executor.Execute(context.Background(), request, operation)
|
||
|
|
if err != nil || !second.Replayed || string(second.Body) != `{"created": true}` {
|
||
|
|
t.Fatalf("second execute = %#v, %v", second, err)
|
||
|
|
}
|
||
|
|
if calls.Load() != 1 {
|
||
|
|
t.Fatalf("operation calls = %d, want 1", calls.Load())
|
||
|
|
}
|
||
|
|
lookedUp, found, err := executor.Lookup(context.Background(), request)
|
||
|
|
if err != nil || !found || !lookedUp.Replayed || string(lookedUp.Body) != `{"created": true}` {
|
||
|
|
t.Fatalf("lookup=%#v found=%v err=%v", lookedUp, found, err)
|
||
|
|
}
|
||
|
|
request.RequestHash = Hash([]byte(`{"value":2}`))
|
||
|
|
if _, err := executor.Execute(context.Background(), request, operation); !errors.Is(err, ErrKeyConflict) {
|
||
|
|
t.Fatalf("hash conflict error = %v, want ErrKeyConflict", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestExecutorSerializesConcurrentRequests(t *testing.T) {
|
||
|
|
pool := integrationPool(t)
|
||
|
|
executor := NewExecutor(pool)
|
||
|
|
key := fmt.Sprintf("concurrent-%d", time.Now().UnixNano())
|
||
|
|
request := Request{
|
||
|
|
ActorKey: "integration-concurrent", Scope: "integration.create", Key: key,
|
||
|
|
RequestHash: Hash([]byte(`{"value":1}`)), ExpiresAfter: time.Hour,
|
||
|
|
}
|
||
|
|
var calls atomic.Int32
|
||
|
|
operation := func(context.Context, pgx.Tx) (Result, error) {
|
||
|
|
calls.Add(1)
|
||
|
|
time.Sleep(30 * time.Millisecond)
|
||
|
|
return Result{StatusCode: 201, Body: json.RawMessage(`{"ok":true}`)}, nil
|
||
|
|
}
|
||
|
|
var waitGroup sync.WaitGroup
|
||
|
|
errorsChannel := make(chan error, 5)
|
||
|
|
for range 5 {
|
||
|
|
waitGroup.Add(1)
|
||
|
|
go func() {
|
||
|
|
defer waitGroup.Done()
|
||
|
|
_, err := executor.Execute(context.Background(), request, operation)
|
||
|
|
errorsChannel <- err
|
||
|
|
}()
|
||
|
|
}
|
||
|
|
waitGroup.Wait()
|
||
|
|
close(errorsChannel)
|
||
|
|
for err := range errorsChannel {
|
||
|
|
if err != nil {
|
||
|
|
t.Errorf("concurrent execute returned %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if calls.Load() != 1 {
|
||
|
|
t.Fatalf("operation calls = %d, want 1", calls.Load())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func integrationPool(t *testing.T) *pgxpool.Pool {
|
||
|
|
t.Helper()
|
||
|
|
databaseURL := os.Getenv("TEST_DATABASE_URL")
|
||
|
|
if databaseURL == "" {
|
||
|
|
t.Skip("TEST_DATABASE_URL is not set")
|
||
|
|
}
|
||
|
|
pool, err := pgxpool.New(context.Background(), databaseURL)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("connect integration database: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(pool.Close)
|
||
|
|
return pool
|
||
|
|
}
|