Nonce Lock

.gitignore

@@ -31,3 +31,4 @@ data/
 bin/
 
 __pycache__/
+.idea/

go.mod

@@ -9,7 +9,6 @@ require (
 	github.com/ethereum/go-ethereum v1.15.11
 	github.com/gin-contrib/cors v1.7.3
 	github.com/jackc/pgx/v5 v5.7.1
-	github.com/joho/godotenv v1.5.1
 	github.com/libp2p/go-libp2p-pubsub v0.13.1
 	github.com/prometheus/client_golang v1.22.0
 	github.com/shutter-network/contracts/v2 v2.0.0-beta.2.0.20250908105003-7e53b1579b04
@@ -102,6 +101,7 @@ require (
 	github.com/koron/go-ssdp v0.0.5 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/libp2p/go-cidranger v1.1.0 // indirect
 	github.com/libp2p/go-flow-metrics v0.2.0 // indirect
@@ -310,6 +310,6 @@ require (
 	github.com/shutter-network/shutter/shlib v0.1.19
 	github.com/swaggo/files v1.0.1
 	golang.org/x/crypto v0.38.0 // indirect
-	golang.org/x/sync v0.14.0 // indirect
+	golang.org/x/sync v0.14.0
 	golang.org/x/text v0.25.0 // indirect
 )

go.sum

@@ -562,8 +562,6 @@ github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPw
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
 github.com/jmhodges/levigo v1.0.0 h1:q5EC36kV79HWeTBWsod3mG11EgStG3qArTKcvlksN1U=
 github.com/jmhodges/levigo v1.0.0/go.mod h1:Q6Qx+uH3RAqyK4rFQroq9RL7mdkABMcfhEI+nNuzMJQ=
-github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
-github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=

internal/service/crypto.go

@@ -248,7 +248,9 @@ func (svc *CryptoService) RegisterIdentity(ctx *gin.Context) {
 		return
 	}
 
-	data, httpErr := svc.CryptoUsecase.RegisterIdentity(ctx, req.DecryptionTimestamp, req.IdentityPrefix)
+	// ctx.Request.Context() rather than ctx: gin.Context's Done channel is nil
+	// unless ContextWithFallback is set, so cancellation would never be seen.
+	data, httpErr := svc.CryptoUsecase.RegisterIdentity(ctx.Request.Context(), req.DecryptionTimestamp, req.IdentityPrefix)
 	if httpErr != nil {
 		ctx.Error(httpErr)
 		return
@@ -420,7 +422,8 @@ func (svc *CryptoService) RegisterEventIdentity(ctx *gin.Context) {
 		return
 	}
 
-	data, httpErr := svc.CryptoUsecase.RegisterEventIdentity(ctx, req.EventTriggerDefinitionHex, req.IdentityPrefix, req.Ttl)
+	// See RegisterIdentity for why this is ctx.Request.Context() and not ctx.
+	data, httpErr := svc.CryptoUsecase.RegisterEventIdentity(ctx.Request.Context(), req.EventTriggerDefinitionHex, req.IdentityPrefix, req.Ttl)
 	if httpErr != nil {
 		ctx.Error(httpErr)
 		return

internal/usecase/crypto.go

@@ -28,9 +28,13 @@ import (
 	httpError "github.com/shutter-network/shutter-api/internal/error"
 	"github.com/shutter-network/shutter-api/metrics"
 	"github.com/shutter-network/shutter/shlib/shcrypto"
+	"golang.org/x/sync/semaphore"
 )
 
-const IdentityPrefixByteLength = 32
+const (
+	IdentityPrefixByteLength            = 32
+	defaultTransactionSubmissionTimeout = 5 * time.Second
+)
 
 type ShutterregistryInterface interface {
 	Registrations(opts *bind.CallOpts, identity [32]byte) (
@@ -92,6 +96,10 @@ type CryptoUsecase struct {
 	keyBroadcastContract         KeyBroadcastInterface
 	ethClient                    EthClientInterface
 	config                       *common.Config
+	// sendSem serializes transaction submission to prevent concurrent requests
+	// producing transactions that use the same nonce.
+	sendSem                      *semaphore.Weighted
+	transactionSubmissionTimeout time.Duration
 }
 
 func NewCryptoUsecase(
@@ -112,6 +120,8 @@ func NewCryptoUsecase(
 		keyBroadcastContract:         keyBroadcastContract,
 		ethClient:                    ethClient,
 		config:                       config,
+		sendSem:                      semaphore.NewWeighted(1),
+		transactionSubmissionTimeout: defaultTransactionSubmissionTimeout,
 	}
 }
 
@@ -120,7 +130,7 @@ func (uc *CryptoUsecase) getSigner(ctx context.Context) (*bind.TransactOpts, *ht
 	chainID, err := uc.ethClient.ChainID(ctx)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying chain id")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying chain id",
 			"",
@@ -143,6 +153,31 @@ func (uc *CryptoUsecase) getSigner(ctx context.Context) (*bind.TransactOpts, *ht
 	return newSigner, nil
 }
 
+// submitTransaction runs submit while holding the submission semaphore, so that
+// only one transaction at a time resolves a nonce and is sent.
+//
+// The request context applies while waiting for the semaphore. Once submission
+// starts, an independent timeout prevents a stalled RPC from holding the
+// semaphore indefinitely.
+func (uc *CryptoUsecase) submitTransaction(ctx context.Context, submit func(context.Context) (*types.Transaction, error)) (*types.Transaction, *httpError.Http, error) {
+	if err := uc.sendSem.Acquire(ctx, 1); err != nil {
+		log.Err(err).Msg("context cancelled while waiting for sendSem")
+		httpErr := httpError.NewHttpError(
+			"internal server error",
+			"",
+			http.StatusInternalServerError,
+		)
+		return nil, &httpErr, nil
+	}
+	defer uc.sendSem.Release(1)
+
+	submitCtx, cancelSubmit := context.WithTimeout(context.Background(), uc.transactionSubmissionTimeout)
+	defer cancelSubmit()
+
+	tx, err := submit(submitCtx)
+	return tx, nil, err
+}
+
 func (uc *CryptoUsecase) GetDecryptionKey(ctx context.Context, identity string) (*GetDecryptionKeyResponse, *httpError.Http) {
 	identityBytes, err := hex.DecodeString(strings.TrimPrefix(string(identity), "0x"))
 	if err != nil {
@@ -168,7 +203,7 @@ func (uc *CryptoUsecase) GetDecryptionKey(ctx context.Context, identity string)
 	registrationData, err := uc.shutterRegistryContract.Registrations(nil, [32]byte(identityBytes))
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying contract")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error while querying for identity from the contract",
 			"",
@@ -301,7 +336,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin
 	blockNumber, err := uc.ethClient.BlockNumber(ctx)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying for recent block")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying for recent block",
 			"",
@@ -313,7 +348,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin
 	eon, err := uc.keyperSetManagerContract.GetKeyperSetIndexByBlock(nil, blockNumber)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying keyper set index")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying for keyper set index",
 			"",
@@ -325,7 +360,7 @@ func (uc *CryptoUsecase) GetDataForEncryption(ctx context.Context, address strin
 	eonKeyBytes, err := uc.keyBroadcastContract.GetEonKey(nil, eon)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying for eon key")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying for eon key",
 			"",
@@ -442,7 +477,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta
 	blockNumber, err := uc.ethClient.BlockNumber(ctx)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying for recent block")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying for recent block",
 			"",
@@ -454,7 +489,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta
 	eon, err := uc.keyperSetManagerContract.GetKeyperSetIndexByBlock(nil, blockNumber)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying keyper set index")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying for keyper set index",
 			"",
@@ -466,7 +501,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta
 	eonKeyBytes, err := uc.keyBroadcastContract.GetEonKey(nil, eon)
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying for eon key")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error encountered while querying for eon key",
 			"",
@@ -496,7 +531,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta
 	registrationData, err := uc.shutterRegistryContract.Registrations(nil, [32]byte(identity))
 	if err != nil {
 		log.Err(err).Msg("err encountered while querying contract")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"error while querying for registrations from the contract",
 			"",
@@ -522,10 +557,16 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta
 		Signer: newSigner.Signer,
 	}
 
-	tx, err := uc.shutterRegistryContract.Register(&opts, eon, identityPrefix, decryptionTimestamp)
+	tx, httpErr, err := uc.submitTransaction(ctx, func(submitCtx context.Context) (*types.Transaction, error) {
+		opts.Context = submitCtx
+		return uc.shutterRegistryContract.Register(&opts, eon, identityPrefix, decryptionTimestamp)
+	})
+	if httpErr != nil {
+		return nil, httpErr
+	}
 	if err != nil {
 		log.Err(err).Msg("failed to send transaction")
-		metrics.TotalFailedRPCCalls.Inc()
+		metrics.FailedRPCCalls.Inc()
 		err := httpError.NewHttpError(
 			"failed to register identity",
 			"",
@@ -537,7 +578,7 @@ func (uc *CryptoUsecase) RegisterIdentity(ctx context.Context, decryptionTimesta
 	// we return the transaction hash in response to allow
 	// users the ability to monitor it themselves
 
-	metrics.TotalSuccessfulIdentityRegistration.Inc()
+	metrics.SuccessfulIdentityRegistrations.Inc()
 	return &RegisterIdentityResponse{
 		Eon:            eon,
 		Identity:       common.PrefixWith0x(hex.EncodeToString(identity)),

internal/usecase/crypto_test.go

@@ -3,14 +3,99 @@ package usecase
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"testing"
+	"time"
 
+	"github.com/ethereum/go-ethereum/core/types"
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
 	"gotest.tools/assert"
 )
 
+func newTransactionSubmissionTestUsecase(timeout time.Duration) *CryptoUsecase {
+	return &CryptoUsecase{
+		sendSem:                      semaphore.NewWeighted(1),
+		transactionSubmissionTimeout: timeout,
+	}
+}
+
+func TestSubmitTransactionSerializesConcurrentSubmissions(t *testing.T) {
+	uc := newTransactionSubmissionTestUsecase(2 * time.Second)
+	const (
+		firstSubmission  = 1
+		secondSubmission = 2
+	)
+
+	transactionSubmissionStarted := make(chan int, 2)
+	finishFirstSubmission := make(chan struct{})
+	var submissions errgroup.Group
+
+	runSubmission := func(submissionID int) error {
+		_, httpErr, err := uc.submitTransaction(context.Background(), func(context.Context) (*types.Transaction, error) {
+			transactionSubmissionStarted <- submissionID
+			if submissionID == firstSubmission {
+				<-finishFirstSubmission
+			}
+			return nil, nil
+		})
+		if httpErr != nil {
+			return fmt.Errorf("submission %d returned HTTP %d", submissionID, httpErr.StatusCode)
+		}
+		if err != nil {
+			return fmt.Errorf("submission %d failed: %w", submissionID, err)
+		}
+		return nil
+	}
+
+	submissions.Go(func() error { return runSubmission(firstSubmission) })
+	assert.Equal(t, <-transactionSubmissionStarted, firstSubmission)
+
+	submissions.Go(func() error { return runSubmission(secondSubmission) })
+	select {
+	case submissionID := <-transactionSubmissionStarted:
+		close(finishFirstSubmission)
+		t.Fatalf("submission %d started while submission 1 was still running", submissionID)
+	case <-time.After(50 * time.Millisecond):
+		// Submission 2 is correctly waiting for submission 1.
+	}
+
+	close(finishFirstSubmission)
+	select {
+	case submissionID := <-transactionSubmissionStarted:
+		assert.Equal(t, submissionID, secondSubmission)
+	case <-time.After(time.Second):
+		t.Fatal("submission 2 did not start after submission 1 finished")
+	}
+
+	assert.NilError(t, submissions.Wait())
+}
+
+func TestSubmitTransactionTimeoutReleasesSemaphore(t *testing.T) {
+	uc := newTransactionSubmissionTestUsecase(50 * time.Millisecond)
+
+	_, httpErr, err := uc.submitTransaction(context.Background(), func(ctx context.Context) (*types.Transaction, error) {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	})
+
+	assert.Assert(t, httpErr == nil)
+	assert.Equal(t, err, context.DeadlineExceeded)
+
+	nextSubmissionStarted := false
+	_, httpErr, err = uc.submitTransaction(context.Background(), func(context.Context) (*types.Transaction, error) {
+		nextSubmissionStarted = true
+		return nil, nil
+	})
+
+	assert.NilError(t, err)
+	assert.Assert(t, httpErr == nil)
+	assert.Assert(t, nextSubmissionStarted)
+}
+
 // Test to verify that QueryExternalKeyper correctly trims quotes and newlines from the response body.
 func TestGetDecryptionKeyFromExternalKeyper_QuotedBody(t *testing.T) {
 	expectedKey := "0xdeadbeefcaffee"