transaction: add NoncePool

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/jackc/pgx/v4 v4.13.0
 	github.com/jdgcs/ed25519 v0.0.0-20200408034030-96c10d46cdc3
 	github.com/jmoiron/sqlx v1.3.4
+	github.com/jonboulle/clockwork v0.4.0
 	github.com/mr-tron/base58 v1.2.0
 	github.com/newrelic/go-agent/v3 v3.20.1
 	github.com/newrelic/go-agent/v3/integrations/nrpgx v1.0.0

go.sum

@@ -386,6 +386,8 @@ github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht
 github.com/jmoiron/sqlx v1.3.4 h1:wv+0IJZfL5z0uZoUjlpKgHkgaFSYD+r9CfrXjEXsO7w=
 github.com/jmoiron/sqlx v1.3.4/go.mod h1:2BljVx/86SuTyjE+aPYlHCTNvZrnJXghYGpNiXLBMCQ=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
+github.com/jonboulle/clockwork v0.4.0 h1:p4Cf1aMWXnXAUh8lVfewRBx1zaTSYKrKMF2g3ST4RZ4=
+github.com/jonboulle/clockwork v0.4.0/go.mod h1:xgRqUGwRcjKCO1vbZUEtSLrqKoPSsUpK7fnezOII0kc=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=

pkg/code/data/internal.go

@@ -180,6 +180,7 @@ type DatabaseData interface {
 	GetNonceCountByStateAndPurpose(ctx context.Context, state nonce.State, purpose nonce.Purpose) (uint64, error)
 	GetAllNonceByState(ctx context.Context, state nonce.State, opts ...query.Option) ([]*nonce.Record, error)
 	GetRandomAvailableNonceByPurpose(ctx context.Context, purpose nonce.Purpose) (*nonce.Record, error)
+	BatchClaimAvailableByPurpose(ctx context.Context, purpose nonce.Purpose, limit int, nodeID string, minExpireAt, maxExpireAt time.Time) ([]*nonce.Record, error)
 	SaveNonce(ctx context.Context, record *nonce.Record) error
 
 	// Fulfillment
@@ -743,6 +744,9 @@ func (dp *DatabaseProvider) GetAllNonceByState(ctx context.Context, state nonce.
 func (dp *DatabaseProvider) GetRandomAvailableNonceByPurpose(ctx context.Context, purpose nonce.Purpose) (*nonce.Record, error) {
 	return dp.nonces.GetRandomAvailableByPurpose(ctx, purpose)
 }
+func (dp *DatabaseProvider) BatchClaimAvailableByPurpose(ctx context.Context, purpose nonce.Purpose, limit int, nodeID string, minExpireAt, maxExpireAt time.Time) ([]*nonce.Record, error) {
+	return dp.nonces.BatchClaimAvailableByPurpose(ctx, purpose, limit, nodeID, minExpireAt, maxExpireAt)
+}
 func (dp *DatabaseProvider) SaveNonce(ctx context.Context, record *nonce.Record) error {
 	return dp.nonces.Save(ctx, record)
 }

pkg/code/data/nonce/memory/store.go

@@ -5,9 +5,10 @@ import (
 	"math/rand"
 	"sort"
 	"sync"
+	"time"
 
-	"github.com/code-payments/code-server/pkg/database/query"
 	"github.com/code-payments/code-server/pkg/code/data/nonce"
+	"github.com/code-payments/code-server/pkg/database/query"
 )
 
 type store struct {
@@ -58,27 +59,27 @@ func (s *store) findAddress(address string) *nonce.Record {
 }
 
 func (s *store) findByState(state nonce.State) []*nonce.Record {
-	res := make([]*nonce.Record, 0)
-	for _, item := range s.records {
-		if item.State == state {
-			res = append(res, item)
-		}
-	}
-	return res
+	return s.findFn(func(nonce *nonce.Record) bool {
+		return nonce.State == state
+	}, -1)
 }
 
 func (s *store) findByStateAndPurpose(state nonce.State, purpose nonce.Purpose) []*nonce.Record {
+	return s.findFn(func(record *nonce.Record) bool {
+		return record.State == state && record.Purpose == purpose
+	}, -1)
+}
+
+func (s *store) findFn(f func(nonce *nonce.Record) bool, limit int) []*nonce.Record {
 	res := make([]*nonce.Record, 0)
 	for _, item := range s.records {
-		if item.State != state {
-			continue
+		if f(item) {
+			res = append(res, item)
 		}
 
-		if item.Purpose != purpose {
-			continue
+		if limit >= 0 && len(res) == limit {
+			break
 		}
-
-		res = append(res, item)
 	}
 	return res
 }
@@ -194,11 +195,47 @@ func (s *store) GetRandomAvailableByPurpose(ctx context.Context, purpose nonce.P
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	items := s.findByStateAndPurpose(nonce.StateAvailable, purpose)
+	items := s.findFn(func(n *nonce.Record) bool {
+		return n.Purpose == purpose && n.IsAvailable()
+	}, -1)
 	if len(items) == 0 {
 		return nil, nonce.ErrNonceNotFound
 	}
 
 	index := rand.Intn(len(items))
 	return items[index], nil
 }
+
+func (s *store) BatchClaimAvailableByPurpose(
+	ctx context.Context,
+	purpose nonce.Purpose,
+	limit int,
+	nodeId string,
+	minExpireAt time.Time,
+	maxExpireAt time.Time,
+) ([]*nonce.Record, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	items := s.findFn(func(n *nonce.Record) bool {
+		return n.Purpose == purpose && n.IsAvailable()
+	}, limit)
+	if len(items) == 0 {
+		return nil, nil
+	}
+
+	for i, l := 0, len(items); i < l; i++ {
+		j := rand.Intn(l)
+		items[i], items[j] = items[j], items[i]
+	}
+	for i := 0; i < len(items); i++ {
+		window := maxExpireAt.Sub(minExpireAt)
+		expiry := minExpireAt.Add(time.Duration(rand.Intn(int(window))))
+
+		items[i].State = nonce.StateClaimed
+		items[i].ClaimNodeId = nodeId
+		items[i].ClaimExpiresAt = expiry
+	}
+
+	return items, nil
+}

pkg/code/data/nonce/nonce.go

@@ -3,6 +3,7 @@ package nonce
 import (
 	"crypto/ed25519"
 	"errors"
+	"time"
 
 	"github.com/mr-tron/base58"
 )
@@ -20,14 +21,11 @@ const (
 	StateAvailable       // The nonce is available to be used by a payment intent, subscription, or other nonce-related transaction.
 	StateReserved        // The nonce is reserved by a payment intent, subscription, or other nonce-related transaction.
 	StateInvalid         // The nonce account is invalid (e.g. insufficient funds, etc).
+	StateClaimed         // The nonce is claimed for future use by a process (identified by Node ID).
 )
 
-// Split nonce pool across different use cases. This has an added benefit of:
-//   - Solving for race conditions without distributed locks.
-//   - Avoiding different use cases from starving each other and ending up in a
-//     deadlocked state. Concretely, it would be really bad if clients could starve
-//     internal processes from creating transactions that would allow us to progress
-//     and submit existing transactions.
+// Purpose indicates the intended use purpose of the nonce. By partitioning nonce's by
+// purpose, we help prevent various use cases from starving each other.
 type Purpose uint8
 
 const (
@@ -46,22 +44,46 @@ type Record struct {
 	Purpose   Purpose
 	State     State
 
+	// Contains the NodeId that transitioned the state into StateClaimed.
+	//
+	// Should be ignored if State != StateClaimed.
+	ClaimNodeId string
+
+	// The time at which StateClaimed is no longer valid, and the state should
+	// be considered StateAvailable.
+	//
+	// Should be ignored if State != StateClaimed.
+	ClaimExpiresAt time.Time
+
 	Signature string
 }
 
 func (r *Record) GetPublicKey() (ed25519.PublicKey, error) {
 	return base58.Decode(r.Address)
 }
 
+func (r *Record) IsAvailable() bool {
+	if r.State == StateAvailable {
+		return true
+	}
+	if r.State != StateClaimed {
+		return false
+	}
+
+	return time.Now().After(r.ClaimExpiresAt)
+}
+
 func (r *Record) Clone() Record {
 	return Record{
-		Id:        r.Id,
-		Address:   r.Address,
-		Authority: r.Authority,
-		Blockhash: r.Blockhash,
-		Purpose:   r.Purpose,
-		State:     r.State,
-		Signature: r.Signature,
+		Id:             r.Id,
+		Address:        r.Address,
+		Authority:      r.Authority,
+		Blockhash:      r.Blockhash,
+		Purpose:        r.Purpose,
+		State:          r.State,
+		ClaimNodeId:    r.ClaimNodeId,
+		ClaimExpiresAt: r.ClaimExpiresAt,
+		Signature:      r.Signature,
 	}
 }
 
@@ -72,21 +94,33 @@ func (r *Record) CopyTo(dst *Record) {
 	dst.Blockhash = r.Blockhash
 	dst.Purpose = r.Purpose
 	dst.State = r.State
+	dst.ClaimNodeId = r.ClaimNodeId
+	dst.ClaimExpiresAt = r.ClaimExpiresAt
 	dst.Signature = r.Signature
 }
 
-func (v *Record) Validate() error {
-	if len(v.Address) == 0 {
+func (r *Record) Validate() error {
+	if len(r.Address) == 0 {
 		return errors.New("nonce account address is required")
 	}
 
-	if len(v.Authority) == 0 {
+	if len(r.Authority) == 0 {
 		return errors.New("authority address is required")
 	}
 
-	if v.Purpose == PurposeUnknown {
+	if r.Purpose == PurposeUnknown {
 		return errors.New("nonce purpose must be set")
 	}
+
+	if r.State == StateClaimed {
+		if r.ClaimNodeId == "" {
+			return errors.New("missing claim node id")
+		}
+		if r.ClaimExpiresAt == (time.Time{}) || r.ClaimExpiresAt.IsZero() {
+			return errors.New("missing claim expiry date")
+		}
+	}
+
 	return nil
 }
 
@@ -102,6 +136,8 @@ func (s State) String() string {
 		return "reserved"
 	case StateInvalid:
 		return "invalid"
+	case StateClaimed:
+		return "claimed"
 	}
 
 	return "unknown"