mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
refactor: move EC2 idempotency adapter to domains/ec2/awsapi
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
// Package gateway_ec2_idem holds the client-token plumbing shared by the EC2
|
||||
// Package idem holds the client-token plumbing shared by the EC2
|
||||
// gateway: the KV bucket every EC2 action's token records live in, and the
|
||||
// extraction of a token and parameter hash from an SDK input struct.
|
||||
package gateway_ec2_idem
|
||||
package idem
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
+8
-8
@@ -1,11 +1,11 @@
|
||||
package gateway_ec2_idem_test
|
||||
package idem_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
gateway_ec2_idem "github.com/mulgadc/spinifex/spinifex/gateway/ec2/idem"
|
||||
ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
@@ -24,7 +24,7 @@ func TestTokenAndParams_ReadsEveryConvertedInput(t *testing.T) {
|
||||
"CreateCapacityReservation": &ec2.CreateCapacityReservationInput{ClientToken: aws.String("tok")},
|
||||
}
|
||||
for action, input := range inputs {
|
||||
token, hash, ok := gateway_ec2_idem.TokenAndParams(input)
|
||||
token, hash, ok := ec2idemapi.TokenAndParams(input)
|
||||
require.True(t, ok, "%s must expose its ClientToken", action)
|
||||
assert.Equal(t, "tok", token, action)
|
||||
assert.NotEmpty(t, hash, action)
|
||||
@@ -43,7 +43,7 @@ func TestTokenAndParams_NoTokenOptsOut(t *testing.T) {
|
||||
"not a struct": aws.String("tok"),
|
||||
}
|
||||
for name, input := range cases {
|
||||
_, _, ok := gateway_ec2_idem.TokenAndParams(input)
|
||||
_, _, ok := ec2idemapi.TokenAndParams(input)
|
||||
assert.False(t, ok, name)
|
||||
}
|
||||
}
|
||||
@@ -56,11 +56,11 @@ func TestTokenAndParams_HashIgnoresTheToken(t *testing.T) {
|
||||
sameParams := &ec2.CreateVolumeInput{Size: aws.Int64(8), ClientToken: aws.String("b")}
|
||||
changed := &ec2.CreateVolumeInput{Size: aws.Int64(16), ClientToken: aws.String("a")}
|
||||
|
||||
_, firstHash, ok := gateway_ec2_idem.TokenAndParams(first)
|
||||
_, firstHash, ok := ec2idemapi.TokenAndParams(first)
|
||||
require.True(t, ok)
|
||||
_, sameHash, ok := gateway_ec2_idem.TokenAndParams(sameParams)
|
||||
_, sameHash, ok := ec2idemapi.TokenAndParams(sameParams)
|
||||
require.True(t, ok)
|
||||
_, changedHash, ok := gateway_ec2_idem.TokenAndParams(changed)
|
||||
_, changedHash, ok := ec2idemapi.TokenAndParams(changed)
|
||||
require.True(t, ok)
|
||||
|
||||
assert.Equal(t, firstHash, sameHash, "the token must not feed the hash")
|
||||
@@ -73,7 +73,7 @@ func TestTokenAndParams_LeavesTheInputIntact(t *testing.T) {
|
||||
t.Parallel()
|
||||
input := &ec2.CreateVolumeInput{Size: aws.Int64(8), ClientToken: aws.String("tok")}
|
||||
|
||||
_, _, ok := gateway_ec2_idem.TokenAndParams(input)
|
||||
_, _, ok := ec2idemapi.TokenAndParams(input)
|
||||
require.True(t, ok)
|
||||
|
||||
require.NotNil(t, input.ClientToken)
|
||||
@@ -8,9 +8,9 @@ import (
|
||||
"sync"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
|
||||
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
|
||||
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
|
||||
gateway_ec2_idem "github.com/mulgadc/spinifex/spinifex/gateway/ec2/idem"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
@@ -48,7 +48,7 @@ func getClientTokenStore(ctx context.Context, nc *nats.Conn) (*ClientTokenStore,
|
||||
// namespaces its keys by action, but RunInstances predates that: re-keying live
|
||||
// records would make a retry re-launch instead of replay.
|
||||
func newClientTokenStore(ctx context.Context, js jetstream.JetStream) (*ClientTokenStore, error) {
|
||||
return idempotency.OpenStore[ec2.Reservation](ctx, js, gateway_ec2_idem.KVBucket, gateway_ec2_idem.TTL)
|
||||
return idempotency.OpenStore[ec2.Reservation](ctx, js, ec2idemapi.KVBucket, ec2idemapi.TTL)
|
||||
}
|
||||
|
||||
// clientTokenParamHash hashes the request excluding ClientToken, so the same
|
||||
|
||||
@@ -7,9 +7,9 @@ import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
|
||||
ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
|
||||
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
|
||||
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
|
||||
gateway_ec2_idem "github.com/mulgadc/spinifex/spinifex/gateway/ec2/idem"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
@@ -27,7 +27,7 @@ func (gw *GatewayConfig) ec2ClientTokenBucket(ctx context.Context) (jetstream.Ke
|
||||
// bucket for every later request. Deadline-free, so the open falls back
|
||||
// to the JetStream API's own timeout.
|
||||
gw.ec2TokenKV, gw.ec2TokenErr = idempotency.OpenBucket(
|
||||
context.WithoutCancel(ctx), js, gateway_ec2_idem.KVBucket, gateway_ec2_idem.TTL)
|
||||
context.WithoutCancel(ctx), js, ec2idemapi.KVBucket, ec2idemapi.TTL)
|
||||
})
|
||||
return gw.ec2TokenKV, gw.ec2TokenErr
|
||||
}
|
||||
@@ -69,7 +69,7 @@ func runEC2Idempotent[In, Out any](
|
||||
var zero Out
|
||||
// Hashed before the handler runs, so a handler that mutates its input cannot
|
||||
// change what a retry of the same token hashes to.
|
||||
token, paramHash, ok := gateway_ec2_idem.TokenAndParams(input)
|
||||
token, paramHash, ok := ec2idemapi.TokenAndParams(input)
|
||||
if !ok {
|
||||
return work()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user