refactor: one paging walk and one client-token store, not six and two (#971)

spinifex/ebsprovider/memory.go

@@ -4,7 +4,8 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"sort"
+	"maps"
+	"slices"
 	"sync"
 	"time"
 )
@@ -162,23 +163,13 @@ func (m *MemoryProvider) ListVolumes(_ context.Context, req ListVolumesRequest)
 
 	m.mu.RLock()
 	defer m.mu.RUnlock()
-	ids := make([]string, 0, len(m.volumes))
-	for id := range m.volumes {
-		if id > req.StartingToken {
-			ids = append(ids, id)
-		}
-	}
-	sort.Strings(ids)
+	ids := slices.Sorted(maps.Keys(m.volumes))
 
-	pageSize := int(req.PageSize())
 	response := &ListVolumesResponse{Versioned: NewVersioned()}
-	if len(ids) > pageSize {
-		response.NextToken = ids[pageSize-1]
-		ids = ids[:pageSize]
-	}
-	for _, id := range ids {
-		response.Volumes = append(response.Volumes, VolumeRef{ID: id, Handle: m.volumes[id].volume.Handle})
-	}
+	response.Volumes, response.NextToken = Page(ids, req.StartingToken, int(req.PageSize()),
+		func(id string) VolumeRef {
+			return VolumeRef{ID: id, Handle: m.volumes[id].volume.Handle}
+		})
 	return response, nil
 }
 
@@ -343,28 +334,17 @@ func (m *MemoryProvider) ListSnapshots(_ context.Context, req ListSnapshotsReque
 
 	m.mu.RLock()
 	defer m.mu.RUnlock()
-	ids := make([]string, 0, len(m.snapshots))
-	for id := range m.snapshots {
-		if id > req.StartingToken {
-			ids = append(ids, id)
-		}
-	}
-	sort.Strings(ids)
+	ids := slices.Sorted(maps.Keys(m.snapshots))
 
-	pageSize := int(req.PageSize())
 	response := &ListSnapshotsResponse{Versioned: NewVersioned()}
-	if len(ids) > pageSize {
-		response.NextToken = ids[pageSize-1]
-		ids = ids[:pageSize]
-	}
-	for _, id := range ids {
-		snapshot := m.snapshots[id]
-		response.Snapshots = append(response.Snapshots, SnapshotRef{
-			ID:             id,
-			SourceVolumeID: snapshot.SourceVolumeID,
-			Handle:         snapshot.Handle,
+	response.Snapshots, response.NextToken = Page(ids, req.StartingToken, int(req.PageSize()),
+		func(id string) SnapshotRef {
+			return SnapshotRef{
+				ID:             id,
+				SourceVolumeID: m.snapshots[id].SourceVolumeID,
+				Handle:         m.snapshots[id].Handle,
+			}
 		})
-	}
 	return response, nil
 }
 

spinifex/ebsprovider/paging.go

@@ -0,0 +1,21 @@
+package ebsprovider
+
+// Page returns one page of refs built from ids, and the token that resumes after
+// it. ids must be sorted ascending; an empty next token means the last page,
+// including when the page fits exactly.
+func Page[T any](ids []string, startingToken string, pageSize int, ref func(id string) T) (page []T, next string) {
+	var last string
+	for _, id := range ids {
+		if id <= startingToken {
+			continue
+		}
+		// Checked before appending, so a token is only issued once an id beyond a
+		// full page is known to exist. An exact fit ends the walk with no token.
+		if len(page) == pageSize {
+			return page, last
+		}
+		page = append(page, ref(id))
+		last = id
+	}
+	return page, ""
+}

spinifex/ebsprovider/paging_test.go

@@ -5,8 +5,85 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/ebsprovider"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
+// ref is the identity builder: these tests are about which ids land on the page
+// and what token follows it, not about the ref type built from them.
+func ref(id string) string { return id }
+
+// The walk each provider used to hand-roll. A token is the last id on the page,
+// so resuming with it skips exactly what was already served.
+func TestPage(t *testing.T) {
+	ids := []string{"a", "b", "c", "d", "e"}
+
+	tests := []struct {
+		name          string
+		ids           []string
+		startingToken string
+		pageSize      int
+		wantPage      []string
+		wantNext      string
+	}{
+		{
+			name: "a short page ends the walk", ids: ids, pageSize: 10,
+			wantPage: ids, wantNext: "",
+		},
+		{
+			name: "an exact fit ends the walk with no token", ids: ids, pageSize: 5,
+			wantPage: ids, wantNext: "",
+		},
+		{
+			name: "a full page with more behind it returns the last id", ids: ids, pageSize: 2,
+			wantPage: []string{"a", "b"}, wantNext: "b",
+		},
+		{
+			name: "resuming skips the token itself", ids: ids, startingToken: "b", pageSize: 2,
+			wantPage: []string{"c", "d"}, wantNext: "d",
+		},
+		{
+			name: "the final page from a token carries no token", ids: ids, startingToken: "c", pageSize: 2,
+			wantPage: []string{"d", "e"}, wantNext: "",
+		},
+		{
+			name: "a token past the end returns nothing", ids: ids, startingToken: "z", pageSize: 2,
+			wantPage: nil, wantNext: "",
+		},
+		{
+			name: "an empty set returns nothing", ids: nil, pageSize: 2,
+			wantPage: nil, wantNext: "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			page, next := ebsprovider.Page(tt.ids, tt.startingToken, tt.pageSize, ref)
+
+			assert.Equal(t, tt.wantPage, page)
+			assert.Equal(t, tt.wantNext, next,
+				"an empty token means last page; a non-empty one must be the last id served")
+		})
+	}
+}
+
+// Paging the whole set one item at a time must serve every id exactly once and
+// terminate, which is the property a resume off-by-one breaks.
+func TestPage_WalkingOneAtATimeServesEveryIDOnce(t *testing.T) {
+	ids := []string{"a", "b", "c", "d", "e"}
+
+	var seen []string
+	token := ""
+	for range len(ids) + 1 {
+		page, next := ebsprovider.Page(ids, token, 1, ref)
+		seen = append(seen, page...)
+		if next == "" {
+			break
+		}
+		token = next
+	}
+
+	require.Equal(t, ids, seen, "every id has to be served exactly once across the walk")
+}
+
 // TestPageSizeClamp pins both list requests' PageSize to the same rule: a
 // MaxResults above the cap is clamped to it, one at or below zero means "the
 // caller has no preference", and anything in between is honoured verbatim.

spinifex/gateway/ec2/instance/clienttoken.go

@@ -2,9 +2,6 @@ package gateway_ec2_instance
 
 import (
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -13,6 +10,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/idempotency"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -24,39 +22,11 @@ const (
 	// clientTokenTTL must outlast SDK retry windows; short enough that a crashed
 	// in-flight record ages out and frees the token for a fresh launch.
 	clientTokenTTL = 15 * time.Minute
-
-	tokenStatusInFlight = "in-flight"
-	tokenStatusDone     = "done"
 )
 
-// clientTokenWaitTimeout caps how long a duplicate caller polls an in-flight winner.
-// clientTokenPollStep is the inter-poll sleep. Vars so tests can shrink them.
-var (
-	clientTokenWaitTimeout = 30 * time.Second
-	clientTokenPollStep    = 250 * time.Millisecond
-)
-
-// errIdempotentParamMismatch signals the same token was reused with different
-// request parameters (AWS IdempotentParameterMismatch).
-var errIdempotentParamMismatch = errors.New("clienttoken: idempotent parameter mismatch")
-
-// errClientTokenWaitTimeout signals a duplicate caller polled an in-flight
-// winner past clientTokenWaitTimeout without the winner finishing.
-var errClientTokenWaitTimeout = errors.New("clienttoken: timed out waiting for in-flight launch")
-
-// tokenRecord is the per-token idempotency record stored in the KV bucket.
-type tokenRecord struct {
-	Status      string           `json:"status"`
-	ParamHash   string           `json:"paramHash"`
-	StartedAt   time.Time        `json:"startedAt"`
-	Reservation *ec2.Reservation `json:"reservation,omitempty"`
-}
-
-// ClientTokenStore implements RunInstances ClientToken idempotency over a TTL KV
-// bucket. The first caller owns the launch; duplicates replay (done) or poll (in-flight).
-type ClientTokenStore struct {
-	kv jetstream.KeyValue
-}
+// ClientTokenStore is the RunInstances idempotency store: the first caller owns
+// the launch, duplicates replay its reservation or poll for it.
+type ClientTokenStore = idempotency.Store[ec2.Reservation]
 
 var (
 	ctStore        *ClientTokenStore
@@ -64,7 +34,9 @@ var (
 	errCTStoreInit error
 )
 
-// getClientTokenStore lazily initialises the process-wide client-token store via sync.Once.
+// getClientTokenStore lazily initialises the process-wide client-token store via
+// sync.Once. Process-wide because RunInstances is a free function with no
+// instance to hang it on.
 func getClientTokenStore(ctx context.Context, nc *nats.Conn) (*ClientTokenStore, error) {
 	ctOnce.Do(func() {
 		js, err := jetstream.New(nc)
@@ -82,22 +54,7 @@ func getClientTokenStore(ctx context.Context, nc *nats.Conn) (*ClientTokenStore,
 }
 
 func newClientTokenStore(ctx context.Context, js jetstream.JetStream) (*ClientTokenStore, error) {
-	kv, err := js.KeyValue(ctx, kvBucketClientTokens)
-	if errors.Is(err, jetstream.ErrBucketNotFound) {
-		kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
-			Bucket:  kvBucketClientTokens,
-			History: 1,
-			TTL:     clientTokenTTL,
-		})
-	}
-	if err != nil {
-		return nil, fmt.Errorf("open client-token bucket: %w", err)
-	}
-	return &ClientTokenStore{kv: kv}, nil
-}
-
-func clientTokenKey(accountID, token string) string {
-	return accountID + "." + token
+	return idempotency.OpenStore[ec2.Reservation](ctx, js, kvBucketClientTokens, clientTokenTTL)
 }
 
 // clientTokenParamHash hashes the request excluding ClientToken, so the same
@@ -105,82 +62,7 @@ func clientTokenKey(accountID, token string) string {
 func clientTokenParamHash(input *ec2.RunInstancesInput) string {
 	clone := *input
 	clone.ClientToken = nil
-	b, err := json.Marshal(&clone)
-	if err != nil {
-		// Fallback: idempotency degrades to name-only.
-		b = []byte(clientTokenKey("", ""))
-	}
-	sum := sha256.Sum256(b)
-	return hex.EncodeToString(sum[:])
-}
-
-// Claim attempts to own the launch for this token.
-// Returns (nil, true, nil) when the caller owns the launch (must Finalize or Abort),
-// (reservation, false, nil) to replay a completed launch, or an error on mismatch/timeout.
-func (c *ClientTokenStore) Claim(ctx context.Context, accountID, token, paramHash string) (*ec2.Reservation, bool, error) {
-	key := clientTokenKey(accountID, token)
-	inflight, err := json.Marshal(tokenRecord{
-		Status:    tokenStatusInFlight,
-		ParamHash: paramHash,
-		StartedAt: time.Now().UTC(),
-	})
-	if err != nil {
-		return nil, false, fmt.Errorf("clienttoken marshal: %w", err)
-	}
-	if _, cerr := c.kv.Create(ctx, key, inflight); cerr == nil {
-		return nil, true, nil
-	} else if !errors.Is(cerr, jetstream.ErrKeyExists) {
-		return nil, false, fmt.Errorf("clienttoken create %s: %w", key, cerr)
-	}
-
-	deadline := time.Now().Add(clientTokenWaitTimeout)
-	for {
-		entry, gerr := c.kv.Get(ctx, key)
-		if gerr != nil {
-			if errors.Is(gerr, jetstream.ErrKeyNotFound) {
-				// Owner aborted or record aged out: race for ownership.
-				if _, rcerr := c.kv.Create(ctx, key, inflight); rcerr == nil {
-					return nil, true, nil
-				} else if !errors.Is(rcerr, jetstream.ErrKeyExists) {
-					return nil, false, fmt.Errorf("clienttoken recreate %s: %w", key, rcerr)
-				}
-				continue
-			}
-			return nil, false, fmt.Errorf("clienttoken get %s: %w", key, gerr)
-		}
-		var rec tokenRecord
-		if uerr := json.Unmarshal(entry.Value(), &rec); uerr != nil {
-			return nil, false, fmt.Errorf("clienttoken unmarshal %s: %w", key, uerr)
-		}
-		if rec.ParamHash != paramHash {
-			return nil, false, errIdempotentParamMismatch
-		}
-		if rec.Status == tokenStatusDone {
-			return rec.Reservation, false, nil
-		}
-		if time.Now().After(deadline) {
-			return nil, false, errClientTokenWaitTimeout
-		}
-		time.Sleep(clientTokenPollStep)
-	}
-}
-
-// Finalize marks the token done with the reservation so duplicates can replay it.
-func (c *ClientTokenStore) Finalize(ctx context.Context, accountID, token, paramHash string, res *ec2.Reservation) error {
-	key := clientTokenKey(accountID, token)
-	data, err := json.Marshal(tokenRecord{
-		Status:      tokenStatusDone,
-		ParamHash:   paramHash,
-		StartedAt:   time.Now().UTC(),
-		Reservation: res,
-	})
-	if err != nil {
-		return fmt.Errorf("clienttoken finalize marshal: %w", err)
-	}
-	if _, err := c.kv.Put(ctx, key, data); err != nil {
-		return fmt.Errorf("clienttoken finalize put %s: %w", key, err)
-	}
-	return nil
+	return idempotency.ParamHash(&clone)
 }
 
 // runInstancesWithClientToken wraps a launch in ClientToken idempotency:
@@ -195,7 +77,7 @@ func runInstancesWithClientToken(
 	var zero ec2.Reservation
 	replay, owned, cerr := store.Claim(ctx, accountID, token, paramHash)
 	if cerr != nil {
-		if errors.Is(cerr, errIdempotentParamMismatch) {
+		if errors.Is(cerr, idempotency.ErrParamMismatch) {
 			return zero, errors.New(awserrors.ErrorIdempotentParameterMismatch)
 		}
 		slog.Error("RunInstances: client-token claim failed", "token", token, "err", cerr)
@@ -218,17 +100,9 @@ func runInstancesWithClientToken(
 		store.Abort(outcomeCtx, accountID, token)
 		return zero, rerr
 	}
-	if ferr := store.Finalize(outcomeCtx, accountID, token, paramHash, &res); ferr != nil {
+	if ferr := store.Finalize(outcomeCtx, accountID, token, paramHash, res); ferr != nil {
 		// Launch succeeded; finalize failure only weakens future dedup — don't fail.
 		slog.Warn("RunInstances: failed to finalize client-token record", "token", token, "err", ferr)
 	}
 	return res, nil
 }
-
-// Abort drops the in-flight token so a retry can re-launch.
-func (c *ClientTokenStore) Abort(ctx context.Context, accountID, token string) {
-	key := clientTokenKey(accountID, token)
-	if err := c.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
-		slog.Warn("clienttoken: failed to abort token record", "key", key, "err", err)
-	}
-}

spinifex/gateway/ec2/instance/clienttoken_test.go

@@ -2,10 +2,7 @@ package gateway_ec2_instance
 
 import (
 	"errors"
-	"sync"
-	"sync/atomic"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -25,103 +22,9 @@ func newTestClientTokenStore(t *testing.T) *ClientTokenStore {
 	return store
 }
 
-// A completed token replays the stored reservation for a duplicate caller with
-// matching params.
-func TestClientToken_ReplaysCompletedReservation(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok, hash = "tok-1", "hash-a"
-
-	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
-	require.NoError(t, err)
-	require.True(t, owned, "first caller owns the launch")
-
-	res := &ec2.Reservation{ReservationId: aws.String("r-123")}
-	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, hash, res))
-
-	replay, owned2, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
-	require.NoError(t, err)
-	assert.False(t, owned2, "duplicate caller does not own the launch")
-	require.NotNil(t, replay)
-	assert.Equal(t, "r-123", aws.StringValue(replay.ReservationId))
-}
-
-// Reusing a token with different params is IdempotentParameterMismatch.
-func TestClientToken_ParamMismatchRejected(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok = "tok-2"
-
-	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, "hash-a")
-	require.NoError(t, err)
-	require.True(t, owned)
-	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, "hash-a", &ec2.Reservation{ReservationId: aws.String("r-1")}))
-
-	_, _, err = store.Claim(t.Context(), ctTestAccount, tok, "hash-DIFFERENT")
-	assert.ErrorIs(t, err, errIdempotentParamMismatch)
-}
-
-// A failed launch aborts the token so a later retry with the same token
-// re-launches instead of replaying a non-existent reservation.
-func TestClientToken_AbortAllowsRelaunch(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok, hash = "tok-3", "hash-a"
-
-	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
-	require.NoError(t, err)
-	require.True(t, owned)
-
-	store.Abort(t.Context(), ctTestAccount, tok)
-
-	_, owned2, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
-	require.NoError(t, err)
-	assert.True(t, owned2, "after abort the token is free to re-own")
-}
-
-// Concurrent callers with the same token must yield exactly one owner; the
-// others replay the owner's reservation. Proves single-launch under -race.
-func TestClientToken_ConcurrentSingleOwner(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok, hash = "tok-4", "hash-a"
-	res := &ec2.Reservation{ReservationId: aws.String("r-only")}
-
-	const n = 4
-	var owners int32
-	var wg sync.WaitGroup
-	errs := make([]error, n)
-	replays := make([]*ec2.Reservation, n)
-	for i := range n {
-		wg.Add(1)
-		go func(i int) {
-			defer wg.Done()
-			replay, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
-			if err != nil {
-				errs[i] = err
-				return
-			}
-			if owned {
-				atomic.AddInt32(&owners, 1)
-				// Simulate the launch then publish the reservation.
-				errs[i] = store.Finalize(t.Context(), ctTestAccount, tok, hash, res)
-				replays[i] = res
-				return
-			}
-			replays[i] = replay
-		}(i)
-	}
-	wg.Wait()
-
-	for _, e := range errs {
-		require.NoError(t, e)
-	}
-	assert.Equal(t, int32(1), owners, "exactly one caller launches")
-	for _, r := range replays {
-		require.NotNil(t, r, "every caller ends with the single reservation")
-		assert.Equal(t, "r-only", aws.StringValue(r.ReservationId))
-	}
-}
+// The claim/replay/abort mechanics live in spinifex/idempotency and are tested
+// there. What is EC2's own is the param hash, the launch orchestration around
+// the store, and the AWS error a mismatch maps onto.
 
 // clientTokenParamHash ignores ClientToken (same params, different token →
 // same hash) but reflects a real parameter change.
@@ -145,7 +48,16 @@ func TestClientTokenParamHash_IgnoresTokenReflectsParams(t *testing.T) {
 		"a real param change must change the hash")
 }
 
-// --- runInstancesWithClientToken (extracted orchestration) ---
+// The hash is taken from a copy, so hashing must not clear the caller's token.
+func TestClientTokenParamHash_LeavesTheInputAlone(t *testing.T) {
+	t.Parallel()
+	input := &ec2.RunInstancesInput{ImageId: aws.String("ami-123"), ClientToken: aws.String("tok-A")}
+
+	clientTokenParamHash(input)
+
+	assert.Equal(t, "tok-A", aws.StringValue(input.ClientToken),
+		"hashing must not strip the token the launch still needs")
+}
 
 // A completed token replays its reservation and must NOT invoke the launcher.
 func TestRunInstancesWithClientToken_ReplaySkipsLaunch(t *testing.T) {
@@ -155,7 +67,7 @@ func TestRunInstancesWithClientToken_ReplaySkipsLaunch(t *testing.T) {
 	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
 	require.NoError(t, err)
 	require.True(t, owned)
-	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, hash, &ec2.Reservation{ReservationId: aws.String("r-x")}))
+	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, hash, ec2.Reservation{ReservationId: aws.String("r-x")}))
 
 	launched := false
 	res, err := runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) {
@@ -218,7 +130,7 @@ func TestRunInstancesWithClientToken_ParamMismatchMapsAWSError(t *testing.T) {
 	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, "hA")
 	require.NoError(t, err)
 	require.True(t, owned)
-	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, "hA", &ec2.Reservation{ReservationId: aws.String("r")}))
+	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, "hA", ec2.Reservation{ReservationId: aws.String("r")}))
 
 	launched := false
 	_, err = runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, "hB", func() (ec2.Reservation, error) {
@@ -241,21 +153,3 @@ func TestGetClientTokenStore_BindsOnce(t *testing.T) {
 	require.NoError(t, err)
 	assert.Same(t, s1, s2, "store binds once")
 }
-
-// A duplicate caller polling an in-flight winner that never finishes bails out
-// with the wait-timeout sentinel (exercises the Claim poll loop + deadline).
-func TestClientToken_InFlightWaitTimesOut(t *testing.T) {
-	store := newTestClientTokenStore(t)
-	const tok, hash = "wt-1", "h"
-
-	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
-	require.NoError(t, err)
-	require.True(t, owned, "owner holds the in-flight record and never finalizes")
-
-	origTimeout, origStep := clientTokenWaitTimeout, clientTokenPollStep
-	clientTokenWaitTimeout, clientTokenPollStep = 30*time.Millisecond, 10*time.Millisecond
-	defer func() { clientTokenWaitTimeout, clientTokenPollStep = origTimeout, origStep }()
-
-	_, _, err = store.Claim(t.Context(), ctTestAccount, tok, hash)
-	assert.ErrorIs(t, err, errClientTokenWaitTimeout)
-}

spinifex/handlers/eks/clienttoken.go

@@ -2,15 +2,11 @@ package handlers_eks
 
 import (
 	"context"
-	"crypto/sha256"
-	"encoding/hex"
-	"encoding/json"
-	"errors"
 	"fmt"
-	"log/slog"
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/mulgadc/spinifex/spinifex/idempotency"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -23,40 +19,12 @@ const (
 	// synchronous phase of CreateCluster, not the async launch: a duplicate
 	// replays the meta by name, kept current by the cluster's own KV record.
 	clusterTokenTTL = 15 * time.Minute
-
-	clusterTokenStatusInFlight = "in-flight"
-	clusterTokenStatusDone     = "done"
 )
 
-// clusterTokenWaitTimeout caps how long a duplicate caller polls an in-flight
-// winner. clusterTokenPollStep is the inter-poll sleep. Vars so tests can shrink them.
-var (
-	clusterTokenWaitTimeout = 30 * time.Second
-	clusterTokenPollStep    = 250 * time.Millisecond
-)
-
-// errClusterTokenParamMismatch signals the same ClientRequestToken was reused
-// with different CreateClusterInput parameters (AWS IdempotentParameterMismatch).
-var errClusterTokenParamMismatch = errors.New("clienttoken: idempotent parameter mismatch")
-
-// errClusterTokenWaitTimeout signals a duplicate caller polled an in-flight
-// winner past clusterTokenWaitTimeout without the winner finishing.
-var errClusterTokenWaitTimeout = errors.New("clienttoken: timed out waiting for in-flight create")
-
-// clusterTokenRecord is the per-token idempotency record stored in the KV bucket.
-type clusterTokenRecord struct {
-	Status      string    `json:"status"`
-	ParamHash   string    `json:"paramHash"`
-	StartedAt   time.Time `json:"startedAt"`
-	ClusterName string    `json:"clusterName,omitempty"`
-}
-
-// ClusterTokenStore implements CreateCluster ClientRequestToken idempotency
-// over a TTL KV bucket, mirroring the EC2 RunInstances ClientToken pattern:
-// the first caller owns the create, duplicates replay or poll.
-type ClusterTokenStore struct {
-	kv jetstream.KeyValue
-}
+// ClusterTokenStore is the CreateCluster idempotency store. It replays the
+// created cluster's name rather than a response, so a duplicate reads the
+// cluster's current state instead of a frozen snapshot.
+type ClusterTokenStore = idempotency.Store[string]
 
 // getClusterTokenStore lazily initialises s's cluster-token store via sync.Once,
 // scoped to the service instance rather than the process, so each EKSServiceImpl
@@ -77,22 +45,7 @@ func (s *EKSServiceImpl) getClusterTokenStore(ctx context.Context) (*ClusterToke
 }
 
 func newClusterTokenStore(ctx context.Context, js jetstream.JetStream) (*ClusterTokenStore, error) {
-	kv, err := js.KeyValue(ctx, kvBucketClusterTokens)
-	if errors.Is(err, jetstream.ErrBucketNotFound) {
-		kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
-			Bucket:  kvBucketClusterTokens,
-			History: 1,
-			TTL:     clusterTokenTTL,
-		})
-	}
-	if err != nil {
-		return nil, fmt.Errorf("open cluster-token bucket: %w", err)
-	}
-	return &ClusterTokenStore{kv: kv}, nil
-}
-
-func clusterTokenKey(accountID, token string) string {
-	return accountID + "." + token
+	return idempotency.OpenStore[string](ctx, js, kvBucketClusterTokens, clusterTokenTTL)
 }
 
 // clusterTokenParamHash hashes the request excluding ClientRequestToken, so
@@ -100,89 +53,5 @@ func clusterTokenKey(accountID, token string) string {
 func clusterTokenParamHash(input *eks.CreateClusterInput) string {
 	clone := *input
 	clone.ClientRequestToken = nil
-	b, err := json.Marshal(&clone)
-	if err != nil {
-		// Fallback: idempotency degrades to name-only.
-		b = []byte(clusterTokenKey("", ""))
-	}
-	sum := sha256.Sum256(b)
-	return hex.EncodeToString(sum[:])
-}
-
-// Claim attempts to own the create for this token.
-// Returns ("", true, nil) when the caller owns the create (must Finalize or Abort),
-// (clusterName, false, nil) to replay a completed create, or an error on mismatch/timeout.
-func (c *ClusterTokenStore) Claim(ctx context.Context, accountID, token, paramHash string) (string, bool, error) {
-	key := clusterTokenKey(accountID, token)
-	inflight, err := json.Marshal(clusterTokenRecord{
-		Status:    clusterTokenStatusInFlight,
-		ParamHash: paramHash,
-		StartedAt: time.Now().UTC(),
-	})
-	if err != nil {
-		return "", false, fmt.Errorf("clustertoken marshal: %w", err)
-	}
-	if _, cerr := c.kv.Create(ctx, key, inflight); cerr == nil {
-		return "", true, nil
-	} else if !errors.Is(cerr, jetstream.ErrKeyExists) {
-		return "", false, fmt.Errorf("clustertoken create %s: %w", key, cerr)
-	}
-
-	deadline := time.Now().Add(clusterTokenWaitTimeout)
-	for {
-		entry, gerr := c.kv.Get(ctx, key)
-		if gerr != nil {
-			if errors.Is(gerr, jetstream.ErrKeyNotFound) {
-				// Owner aborted or record aged out: race for ownership.
-				if _, rcerr := c.kv.Create(ctx, key, inflight); rcerr == nil {
-					return "", true, nil
-				} else if !errors.Is(rcerr, jetstream.ErrKeyExists) {
-					return "", false, fmt.Errorf("clustertoken recreate %s: %w", key, rcerr)
-				}
-				continue
-			}
-			return "", false, fmt.Errorf("clustertoken get %s: %w", key, gerr)
-		}
-		var rec clusterTokenRecord
-		if uerr := json.Unmarshal(entry.Value(), &rec); uerr != nil {
-			return "", false, fmt.Errorf("clustertoken unmarshal %s: %w", key, uerr)
-		}
-		if rec.ParamHash != paramHash {
-			return "", false, errClusterTokenParamMismatch
-		}
-		if rec.Status == clusterTokenStatusDone {
-			return rec.ClusterName, false, nil
-		}
-		if time.Now().After(deadline) {
-			return "", false, errClusterTokenWaitTimeout
-		}
-		time.Sleep(clusterTokenPollStep)
-	}
-}
-
-// Finalize marks the token done with the created cluster's name, so duplicates
-// replay its current state via GetClusterMeta rather than a frozen snapshot.
-func (c *ClusterTokenStore) Finalize(ctx context.Context, accountID, token, paramHash, clusterName string) error {
-	key := clusterTokenKey(accountID, token)
-	data, err := json.Marshal(clusterTokenRecord{
-		Status:      clusterTokenStatusDone,
-		ParamHash:   paramHash,
-		StartedAt:   time.Now().UTC(),
-		ClusterName: clusterName,
-	})
-	if err != nil {
-		return fmt.Errorf("clustertoken finalize marshal: %w", err)
-	}
-	if _, err := c.kv.Put(ctx, key, data); err != nil {
-		return fmt.Errorf("clustertoken finalize put %s: %w", key, err)
-	}
-	return nil
-}
-
-// Abort drops the in-flight token so a retry can re-attempt the create.
-func (c *ClusterTokenStore) Abort(ctx context.Context, accountID, token string) {
-	key := clusterTokenKey(accountID, token)
-	if err := c.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
-		slog.Warn("clienttoken: failed to abort token record", "key", key, "err", err)
-	}
+	return idempotency.ParamHash(&clone)
 }

spinifex/handlers/eks/clienttoken_test.go

@@ -0,0 +1,67 @@
+package handlers_eks
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The store mechanics are tested in spinifex/idempotency. What is EKS's own is
+// the param hash and that the store binds against the cluster-token bucket.
+
+// clusterTokenParamHash ignores ClientRequestToken (same params, different token
+// → same hash) but reflects a real parameter change.
+func TestClusterTokenParamHash_IgnoresTokenReflectsParams(t *testing.T) {
+	t.Parallel()
+	base := &eks.CreateClusterInput{
+		Name:               aws.String("c1"),
+		Version:            aws.String("1.32"),
+		RoleArn:            aws.String("arn:aws:iam::000000000000:role/eksrole"),
+		ClientRequestToken: aws.String("tok-A"),
+	}
+	sameParamsDiffToken := *base
+	sameParamsDiffToken.ClientRequestToken = aws.String("tok-B")
+	diffParams := *base
+	diffParams.Version = aws.String("1.31")
+
+	assert.Equal(t, clusterTokenParamHash(base), clusterTokenParamHash(&sameParamsDiffToken),
+		"token must not affect the param hash")
+	assert.NotEqual(t, clusterTokenParamHash(base), clusterTokenParamHash(&diffParams),
+		"a real param change must change the hash")
+}
+
+// The hash is taken from a copy, so hashing must not clear the caller's token.
+func TestClusterTokenParamHash_LeavesTheInputAlone(t *testing.T) {
+	t.Parallel()
+	input := &eks.CreateClusterInput{Name: aws.String("c1"), ClientRequestToken: aws.String("tok-A")}
+
+	clusterTokenParamHash(input)
+
+	assert.Equal(t, "tok-A", aws.StringValue(input.ClientRequestToken),
+		"hashing must not strip the token the create still needs")
+}
+
+// A cluster name round-trips through the store, which is what a duplicate
+// CreateCluster resolves its reply from.
+func TestClusterTokenStore_ReplaysTheClusterName(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	store, err := newClusterTokenStore(t.Context(), testutil.NewJetStream(t, nc))
+	require.NoError(t, err)
+	const account, tok, hash = "111122223333", "tok-1", "h"
+
+	_, owned, err := store.Claim(t.Context(), account, tok, hash)
+	require.NoError(t, err)
+	require.True(t, owned)
+	require.NoError(t, store.Finalize(t.Context(), account, tok, hash, "my-cluster"))
+
+	replay, owned2, err := store.Claim(t.Context(), account, tok, hash)
+	require.NoError(t, err)
+	assert.False(t, owned2)
+	require.NotNil(t, replay)
+	assert.Equal(t, "my-cluster", *replay)
+}

spinifex/handlers/eks/service_impl.go

@@ -26,6 +26,7 @@ import (
 	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
 	"github.com/mulgadc/spinifex/spinifex/handlers/sysinstance"
+	"github.com/mulgadc/spinifex/spinifex/idempotency"
 	"github.com/mulgadc/spinifex/spinifex/kvlease"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -540,13 +541,19 @@ func (s *EKSServiceImpl) CreateCluster(ctx context.Context, input *eks.CreateClu
 		tokenHash = clusterTokenParamHash(input)
 		replayName, owned, cerr := tokenStore.Claim(ctx, accountID, token, tokenHash)
 		if cerr != nil {
-			if errors.Is(cerr, errClusterTokenParamMismatch) {
+			if errors.Is(cerr, idempotency.ErrParamMismatch) {
 				return nil, errors.New(awserrors.ErrorIdempotentParameterMismatch)
 			}
 			return nil, logCreateErr(name, accountID, "client token claim", cerr)
 		}
 		if !owned {
-			replayMeta, gerr := GetClusterMeta(ctx, acctKV, replayName)
+			// A finalized record always carries the name; without one there is
+			// nothing to replay and re-creating would defeat the token.
+			if replayName == nil {
+				return nil, logCreateErr(name, accountID, "client token replay",
+					errors.New("token record carries no cluster name"))
+			}
+			replayMeta, gerr := GetClusterMeta(ctx, acctKV, *replayName)
 			if gerr != nil {
 				return nil, logCreateErr(name, accountID, "client token replay", gerr)
 			}

spinifex/idempotency/store.go

@@ -0,0 +1,168 @@
+// Package idempotency implements request idempotency over a TTL KV bucket: the
+// first caller to present a token owns the work, and duplicates replay its
+// result or wait for it. The payload a replay returns is the caller's own type.
+package idempotency
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	statusInFlight = "in-flight"
+	statusDone     = "done"
+)
+
+// waitTimeout caps how long a duplicate caller polls an in-flight owner.
+// pollStep is the inter-poll sleep. Vars so tests can shrink them.
+var (
+	waitTimeout = 30 * time.Second
+	pollStep    = 250 * time.Millisecond
+)
+
+// ErrParamMismatch signals a token reused with different request parameters.
+// ErrWaitTimeout signals a duplicate polled an in-flight owner past WaitTimeout.
+var (
+	ErrParamMismatch = errors.New("idempotency: parameter mismatch")
+	ErrWaitTimeout   = errors.New("idempotency: timed out waiting for the in-flight owner")
+)
+
+// record is the per-token record stored in the KV bucket.
+type record[T any] struct {
+	Status    string    `json:"status"`
+	ParamHash string    `json:"paramHash"`
+	StartedAt time.Time `json:"startedAt"`
+	Payload   *T        `json:"payload,omitempty"`
+}
+
+// Store owns the token records for one kind of request. T is what a duplicate
+// caller replays: a whole response, or just the name of what was created.
+type Store[T any] struct {
+	kv jetstream.KeyValue
+}
+
+// OpenStore binds the bucket, creating it with the given TTL if absent. The
+// caller owns the JetStream handle, so a store never outlives the connection it
+// was built from.
+func OpenStore[T any](ctx context.Context, js jetstream.JetStream, bucket string, ttl time.Duration) (*Store[T], error) {
+	kv, err := js.KeyValue(ctx, bucket)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+			Bucket:  bucket,
+			History: 1,
+			TTL:     ttl,
+		})
+	}
+	if err != nil {
+		return nil, fmt.Errorf("open idempotency bucket %s: %w", bucket, err)
+	}
+	return &Store[T]{kv: kv}, nil
+}
+
+// Key scopes a token to its account, so the same token from two accounts is two
+// separate pieces of work.
+func Key(accountID, token string) string {
+	return accountID + "." + token
+}
+
+// ParamHash hashes a request so a token reused with different parameters is
+// caught. The caller must clear the token field first, so identical requests
+// always hash identically.
+func ParamHash(request any) string {
+	b, err := json.Marshal(request)
+	if err != nil {
+		// Fallback: idempotency degrades to token-only.
+		b = []byte{}
+	}
+	sum := sha256.Sum256(b)
+	return hex.EncodeToString(sum[:])
+}
+
+// Claim attempts to own the work for this token. It returns (nil, true, nil)
+// when the caller owns it and must Finalize or Abort, (payload, false, nil) to
+// replay a completed one, or an error on mismatch or timeout.
+func (s *Store[T]) Claim(ctx context.Context, accountID, token, paramHash string) (*T, bool, error) {
+	key := Key(accountID, token)
+	inflight, err := json.Marshal(record[T]{
+		Status:    statusInFlight,
+		ParamHash: paramHash,
+		StartedAt: time.Now().UTC(),
+	})
+	if err != nil {
+		return nil, false, fmt.Errorf("idempotency marshal: %w", err)
+	}
+	if _, cerr := s.kv.Create(ctx, key, inflight); cerr == nil {
+		return nil, true, nil
+	} else if !errors.Is(cerr, jetstream.ErrKeyExists) {
+		return nil, false, fmt.Errorf("idempotency create %s: %w", key, cerr)
+	}
+
+	deadline := time.Now().Add(waitTimeout)
+	for {
+		entry, gerr := s.kv.Get(ctx, key)
+		if gerr != nil {
+			if errors.Is(gerr, jetstream.ErrKeyNotFound) {
+				// Owner aborted or the record aged out: race for ownership.
+				if _, rcerr := s.kv.Create(ctx, key, inflight); rcerr == nil {
+					return nil, true, nil
+				} else if !errors.Is(rcerr, jetstream.ErrKeyExists) {
+					return nil, false, fmt.Errorf("idempotency recreate %s: %w", key, rcerr)
+				}
+				continue
+			}
+			return nil, false, fmt.Errorf("idempotency get %s: %w", key, gerr)
+		}
+		var rec record[T]
+		if uerr := json.Unmarshal(entry.Value(), &rec); uerr != nil {
+			return nil, false, fmt.Errorf("idempotency unmarshal %s: %w", key, uerr)
+		}
+		if rec.ParamHash != paramHash {
+			return nil, false, ErrParamMismatch
+		}
+		if rec.Status == statusDone {
+			return rec.Payload, false, nil
+		}
+		if time.Now().After(deadline) {
+			return nil, false, ErrWaitTimeout
+		}
+		time.Sleep(pollStep)
+	}
+}
+
+// Finalize marks the token done with the payload duplicates should replay.
+func (s *Store[T]) Finalize(ctx context.Context, accountID, token, paramHash string, payload T) error {
+	key := Key(accountID, token)
+	data, err := json.Marshal(record[T]{
+		Status:    statusDone,
+		ParamHash: paramHash,
+		StartedAt: time.Now().UTC(),
+		Payload:   &payload,
+	})
+	if err != nil {
+		return fmt.Errorf("idempotency finalize marshal: %w", err)
+	}
+	if _, err := s.kv.Put(ctx, key, data); err != nil {
+		return fmt.Errorf("idempotency finalize put %s: %w", key, err)
+	}
+	return nil
+}
+
+// Abort drops the in-flight token so a retry can re-attempt the work.
+func (s *Store[T]) Abort(ctx context.Context, accountID, token string) {
+	key := Key(accountID, token)
+	if err := s.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		slog.Warn("idempotency: failed to abort token record", "key", key, "err", err)
+	}
+}
+
+// There is deliberately no Do wrapper. The two callers orchestrate differently:
+// RunInstances wraps one launch closure, while CreateCluster interleaves aborts
+// through a long create and replays by name rather than from the payload.

spinifex/idempotency/store_test.go

@@ -0,0 +1,174 @@
+package idempotency_test
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/idempotency"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const testAccount = "111122223333"
+
+// payload stands in for whatever a caller replays. A struct rather than a string
+// so the JSON round-trip through the record is actually exercised.
+type payload struct {
+	ID string `json:"id"`
+}
+
+func newTestStore[T any](t *testing.T) *idempotency.Store[T] {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	store, err := idempotency.OpenStore[T](t.Context(), testutil.NewJetStream(t, nc), "test-tokens", time.Minute)
+	require.NoError(t, err)
+	return store
+}
+
+// A completed token replays its payload to a duplicate caller with matching
+// params, which is the whole point of the store.
+func TestStore_ReplaysCompletedPayload(t *testing.T) {
+	t.Parallel()
+	store := newTestStore[payload](t)
+	const tok, hash = "tok-1", "hash-a"
+
+	_, owned, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	require.True(t, owned, "first caller owns the work")
+
+	require.NoError(t, store.Finalize(t.Context(), testAccount, tok, hash, payload{ID: "r-123"}))
+
+	replay, owned2, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	assert.False(t, owned2, "duplicate caller does not own the work")
+	require.NotNil(t, replay)
+	assert.Equal(t, "r-123", replay.ID)
+}
+
+// The payload type is the caller's: EKS replays a bare cluster name, so a
+// non-struct T has to round-trip too.
+func TestStore_ReplaysAStringPayload(t *testing.T) {
+	t.Parallel()
+	store := newTestStore[string](t)
+	const tok, hash = "tok-str", "hash-a"
+
+	_, owned, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	require.True(t, owned)
+	require.NoError(t, store.Finalize(t.Context(), testAccount, tok, hash, "my-cluster"))
+
+	replay, _, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	require.NotNil(t, replay)
+	assert.Equal(t, "my-cluster", *replay)
+}
+
+// Reusing a token with different params is the mismatch case, which each caller
+// maps onto IdempotentParameterMismatch.
+func TestStore_ParamMismatchRejected(t *testing.T) {
+	t.Parallel()
+	store := newTestStore[payload](t)
+	const tok = "tok-2"
+
+	_, owned, err := store.Claim(t.Context(), testAccount, tok, "hash-a")
+	require.NoError(t, err)
+	require.True(t, owned)
+	require.NoError(t, store.Finalize(t.Context(), testAccount, tok, "hash-a", payload{ID: "r-1"}))
+
+	_, _, err = store.Claim(t.Context(), testAccount, tok, "hash-DIFFERENT")
+	assert.ErrorIs(t, err, idempotency.ErrParamMismatch)
+}
+
+// A failed attempt aborts the token so a later retry re-runs the work instead of
+// replaying a result that was never produced.
+func TestStore_AbortAllowsRetry(t *testing.T) {
+	t.Parallel()
+	store := newTestStore[payload](t)
+	const tok, hash = "tok-3", "hash-a"
+
+	_, owned, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	require.True(t, owned)
+
+	store.Abort(t.Context(), testAccount, tok)
+
+	_, owned2, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	assert.True(t, owned2, "after abort the token is free to re-own")
+}
+
+// Concurrent callers with the same token must yield exactly one owner and the
+// rest replay it. This is the property that stops a double launch under -race.
+func TestStore_ConcurrentSingleOwner(t *testing.T) {
+	t.Parallel()
+	store := newTestStore[payload](t)
+	const tok, hash = "tok-4", "hash-a"
+	done := payload{ID: "r-only"}
+
+	const n = 4
+	var owners int32
+	var wg sync.WaitGroup
+	errs := make([]error, n)
+	replays := make([]*payload, n)
+	for i := range n {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			replay, owned, err := store.Claim(t.Context(), testAccount, tok, hash)
+			if err != nil {
+				errs[i] = err
+				return
+			}
+			if owned {
+				atomic.AddInt32(&owners, 1)
+				errs[i] = store.Finalize(t.Context(), testAccount, tok, hash, done)
+				replays[i] = &done
+				return
+			}
+			replays[i] = replay
+		}(i)
+	}
+	wg.Wait()
+
+	for _, e := range errs {
+		require.NoError(t, e)
+	}
+	assert.Equal(t, int32(1), owners, "exactly one caller runs the work")
+	for _, r := range replays {
+		require.NotNil(t, r, "every caller ends with the single result")
+		assert.Equal(t, "r-only", r.ID)
+	}
+}
+
+// The same token under two accounts is two separate pieces of work, so one
+// account's result must never be replayed to another.
+func TestStore_TokensAreScopedPerAccount(t *testing.T) {
+	t.Parallel()
+	store := newTestStore[payload](t)
+	const tok, hash = "shared-token", "h"
+
+	_, owned, err := store.Claim(t.Context(), testAccount, tok, hash)
+	require.NoError(t, err)
+	require.True(t, owned)
+	require.NoError(t, store.Finalize(t.Context(), testAccount, tok, hash, payload{ID: "r-first"}))
+
+	replay, owned2, err := store.Claim(t.Context(), "999988887777", tok, hash)
+	require.NoError(t, err)
+	assert.True(t, owned2, "a second account owns its own work under the same token")
+	assert.Nil(t, replay, "one account must never replay another's result")
+}
+
+// ParamHash reflects the request it is given, which is what makes a reused token
+// with changed parameters detectable.
+func TestParamHash_DistinguishesRequests(t *testing.T) {
+	t.Parallel()
+	first := idempotency.ParamHash(payload{ID: "a"})
+	again := idempotency.ParamHash(payload{ID: "a"})
+	changed := idempotency.ParamHash(payload{ID: "b"})
+
+	assert.Equal(t, first, again, "the same request must always hash the same")
+	assert.NotEqual(t, first, changed, "a changed request must change the hash")
+}

spinifex/idempotency/wait_test.go

@@ -0,0 +1,33 @@
+//test:in-package — the poll deadline is unexported on purpose, so shrinking it
+//to keep the test fast needs to happen from inside the package.
+
+package idempotency
+
+import (
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A duplicate polling an owner that never finishes has to give up rather than
+// block the request forever.
+func TestStore_InFlightWaitTimesOut(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	store, err := OpenStore[string](t.Context(), testutil.NewJetStream(t, nc), "wait-tokens", time.Minute)
+	require.NoError(t, err)
+	const account, tok, hash = "111122223333", "wt-1", "h"
+
+	_, owned, err := store.Claim(t.Context(), account, tok, hash)
+	require.NoError(t, err)
+	require.True(t, owned, "owner holds the in-flight record and never finalizes")
+
+	origTimeout, origStep := waitTimeout, pollStep
+	waitTimeout, pollStep = 30*time.Millisecond, 10*time.Millisecond
+	defer func() { waitTimeout, pollStep = origTimeout, origStep }()
+
+	_, _, err = store.Claim(t.Context(), account, tok, hash)
+	assert.ErrorIs(t, err, ErrWaitTimeout)
+}

spinifex/services/qemunbdd/provider.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -349,25 +350,28 @@ func (p *Provider) ListVolumes(_ context.Context, req ebsprovider.ListVolumesReq
 		return nil, fmt.Errorf("qemunbdd: read volumes directory: %w", err)
 	}
 
-	pageSize := int(req.PageSize())
 	response := &ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned()}
-	// ReadDir returns entries sorted by filename, so resuming after the token
-	// walks the same order the previous page ended on.
+	response.Volumes, response.NextToken = ebsprovider.Page(qcow2IDs(entries), req.StartingToken, int(req.PageSize()),
+		func(id string) ebsprovider.VolumeRef {
+			return ebsprovider.VolumeRef{ID: id, Handle: p.volumePath(id)}
+		})
+	return response, nil
+}
+
+// qcow2IDs returns the sorted resource IDs backing a directory of qcow2 files.
+// Sorted on the trimmed ID rather than the filename: trimming ".qcow2" can
+// reorder entries, since '-' sorts before '.'.
+func qcow2IDs(entries []os.DirEntry) []string {
+	ids := make([]string, 0, len(entries))
 	for _, entry := range entries {
-		volumeID := strings.TrimSuffix(entry.Name(), ".qcow2")
-		if entry.IsDir() || volumeID == entry.Name() || volumeID <= req.StartingToken {
+		id := strings.TrimSuffix(entry.Name(), ".qcow2")
+		if entry.IsDir() || id == entry.Name() {
 			continue
 		}
-		if len(response.Volumes) == pageSize {
-			response.NextToken = response.Volumes[pageSize-1].ID
-			break
-		}
-		response.Volumes = append(response.Volumes, ebsprovider.VolumeRef{
-			ID:     volumeID,
-			Handle: p.volumePath(volumeID),
-		})
+		ids = append(ids, id)
 	}
-	return response, nil
+	slices.Sort(ids)
+	return ids
 }
 
 func (p *Provider) ExpandVolume(ctx context.Context, req ebsprovider.ExpandVolumeRequest) (*ebsprovider.Volume, error) {
@@ -523,25 +527,15 @@ func (p *Provider) ListSnapshots(_ context.Context, req ebsprovider.ListSnapshot
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	pageSize := int(req.PageSize())
 	response := &ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned()}
-	// ReadDir returns entries sorted by filename, so resuming after the token
-	// walks the same order the previous page ended on.
-	for _, entry := range entries {
-		snapshotID := strings.TrimSuffix(entry.Name(), ".qcow2")
-		if entry.IsDir() || snapshotID == entry.Name() || snapshotID <= req.StartingToken {
-			continue
-		}
-		if len(response.Snapshots) == pageSize {
-			response.NextToken = response.Snapshots[pageSize-1].ID
-			break
-		}
-		response.Snapshots = append(response.Snapshots, ebsprovider.SnapshotRef{
-			ID:             snapshotID,
-			SourceVolumeID: p.snapshotMeta[snapshotID].sourceVolumeID,
-			Handle:         p.snapshotPath(snapshotID),
+	response.Snapshots, response.NextToken = ebsprovider.Page(qcow2IDs(entries), req.StartingToken, int(req.PageSize()),
+		func(id string) ebsprovider.SnapshotRef {
+			return ebsprovider.SnapshotRef{
+				ID:             id,
+				SourceVolumeID: p.snapshotMeta[id].sourceVolumeID,
+				Handle:         p.snapshotPath(id),
+			}
 		})
-	}
 	return response, nil
 }
 

spinifex/services/viperblockd/provider_handlers.go

@@ -353,17 +353,10 @@ func handleListVolumes(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	}
 
 	response := ebsprovider.ListVolumesResponse{Versioned: ebsprovider.NewVersioned()}
-	pageSize := int(req.PageSize())
-	for _, id := range ids {
-		if id <= req.StartingToken {
-			continue
-		}
-		if len(response.Volumes) == pageSize {
-			response.NextToken = response.Volumes[pageSize-1].ID
-			break
-		}
-		response.Volumes = append(response.Volumes, ebsprovider.VolumeRef{ID: id, Handle: volumeHandle(id)})
-	}
+	response.Volumes, response.NextToken = ebsprovider.Page(ids, req.StartingToken, int(req.PageSize()),
+		func(id string) ebsprovider.VolumeRef {
+			return ebsprovider.VolumeRef{ID: id, Handle: volumeHandle(id)}
+		})
 	respondProvider(ctx, msg, response)
 }
 
@@ -390,17 +383,10 @@ func handleListSnapshots(ctx context.Context, cfg *Config, msg *nats.Msg) {
 	}
 
 	response := ebsprovider.ListSnapshotsResponse{Versioned: ebsprovider.NewVersioned()}
-	pageSize := int(req.PageSize())
-	for _, id := range ids {
-		if id <= req.StartingToken {
-			continue
-		}
-		if len(response.Snapshots) == pageSize {
-			response.NextToken = response.Snapshots[pageSize-1].ID
-			break
-		}
-		response.Snapshots = append(response.Snapshots, ebsprovider.SnapshotRef{ID: id, Handle: snapshotHandle(id)})
-	}
+	response.Snapshots, response.NextToken = ebsprovider.Page(ids, req.StartingToken, int(req.PageSize()),
+		func(id string) ebsprovider.SnapshotRef {
+			return ebsprovider.SnapshotRef{ID: id, Handle: snapshotHandle(id)}
+		})
 	respondProvider(ctx, msg, response)
 }