diff --git a/cmds/core-service/main.go b/cmds/core-service/main.go index 9c1959869..6b4026fda 100644 --- a/cmds/core-service/main.go +++ b/cmds/core-service/main.go @@ -26,6 +26,7 @@ import ( dsserr "github.com/interuss/dss/pkg/errors" requestlocality "github.com/interuss/dss/pkg/locality" "github.com/interuss/dss/pkg/logging" + "github.com/interuss/dss/pkg/random" "github.com/interuss/dss/pkg/rid/application" rid_v1 "github.com/interuss/dss/pkg/rid/server/v1" rid_v2 "github.com/interuss/dss/pkg/rid/server/v2" @@ -368,6 +369,7 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st handler = http.TimeoutHandler(handler, *timeout, "request timeout") handler = logging.HTTPMiddleware(logger, *dumpRequests, handler) handler = timestamp.RequestTimestampMiddleware(handler) + handler = random.Middleware(handler) handler = requestlocality.LocalityMiddleware(locality)(handler) if *enableMetrics || *enableTracing { diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go index eb002ced6..bb63fda74 100644 --- a/pkg/raftstore/consensus/proposal.go +++ b/pkg/raftstore/consensus/proposal.go @@ -6,6 +6,7 @@ import ( "time" "github.com/google/uuid" + "github.com/interuss/dss/pkg/random" "github.com/interuss/dss/pkg/timestamp" ) @@ -17,10 +18,13 @@ type EntryCommit struct { } type Proposal struct { - ID string `json:"id"` - Locality string `json:"locality"` - NodeID uint64 `json:"node_id"` - Timestamp time.Time `json:"timestamp"` + ID string `json:"id"` + Locality string `json:"locality"` + NodeID uint64 `json:"node_id"` + Timestamp time.Time `json:"timestamp"` + // Seed is the request's random seed (see pkg/random). It is generated by the proposing node + // so that every node deriving pseudo-random data (e.g. generated UUIDs) while applying this proposal derives the exact same values. + Seed int64 `json:"seed"` RequestType RequestType `json:"request_type"` Value []byte `json:"value"` // ReadOnly proposals do not modify the state machine and, @@ -32,12 +36,14 @@ type Proposal struct { func (c *Consensus) newProposal(ctx context.Context, requestType RequestType, value []byte, readOnly bool) Proposal { timestamp := timestamp.MustGetRequestTimestamp(ctx) + seed := random.MustFromContext(ctx) return Proposal{ ID: uuid.NewString(), Locality: c.locality, NodeID: c.nodeID, Timestamp: timestamp.UTC(), + Seed: seed, RequestType: requestType, Value: value, ReadOnly: readOnly, diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index 83ef22b33..a17c00827 100644 --- a/pkg/raftstore/store.go +++ b/pkg/raftstore/store.go @@ -7,6 +7,7 @@ import ( "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/raftstore/consensus" raftparams "github.com/interuss/dss/pkg/raftstore/params" + "github.com/interuss/dss/pkg/random" "github.com/interuss/dss/pkg/store" "github.com/interuss/dss/pkg/timestamp" "github.com/interuss/stacktrace" @@ -117,6 +118,7 @@ func (s *Store[R]) processCommits(ctx context.Context, commitCh <-chan consensus proposalCtx := timestamp.WithRequestTimestamp(ctx, commit.Prop.Timestamp) proposalCtx = locality.WithRequestLocality(proposalCtx, commit.Prop.Locality) + proposalCtx = random.NewContext(proposalCtx, commit.Prop.Seed) result, err := s.raftRepo.Apply(proposalCtx, commit.Prop) commit.Done <- consensus.ProposalResult{Result: result, Error: err} } diff --git a/pkg/random/random.go b/pkg/random/random.go new file mode 100644 index 000000000..99b20dc81 --- /dev/null +++ b/pkg/random/random.go @@ -0,0 +1,76 @@ +// Package random provides a per-request seed. +// +// Used by business logic operations that need to generate pseudo-random data (e.g. UUIDs) +// while still executing deterministically (every Raft node / replay must derive the same value). +package random + +import ( + "context" + "crypto/rand" + "encoding/binary" + "hash/fnv" + mrand "math/rand" + "net/http" + + "github.com/interuss/stacktrace" +) + +type key struct{} + +func newSeed() (int64, error) { + var buf [8]byte + _, err := rand.Read(buf[:]) + if err != nil { + return 0, stacktrace.Propagate(err, "failed to generate random seed") + } + return int64(binary.BigEndian.Uint64(buf[:])), nil +} + +// MustFromContext returns the request seed from the context and panics if it is not present, +// which is a programming error. +func MustFromContext(ctx context.Context) int64 { + seed, ok := ctx.Value(key{}).(int64) + if !ok { + panic(stacktrace.NewError("seed not found in context")) + } + return seed +} + +// NewContext returns a new context with the given seed. +func NewContext(ctx context.Context, seed int64) context.Context { + return context.WithValue(ctx, key{}, seed) +} + +// Generator deterministically builds a pseudo-random generator from the given seed and label. +// The same (seed, label) pair always yields a generator producing the same sequence of values. +// Distinct labels let a single request derive multiple, independent values. +func Generator(seed int64, label string) (*mrand.Rand, error) { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], uint64(seed)) + + h := fnv.New64a() + _, err := h.Write(buf[:]) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to hash seed") + } + _, err = h.Write([]byte(label)) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to hash label") + } + + return mrand.New(mrand.NewSource(int64(h.Sum64()))), nil +} + +// Middleware is an HTTP middleware that stamps each incoming request with a fresh seed, so that +// any pseudo-random data (e.g. generated UUIDs) needed while executing it can be derived +// deterministically via Generator. +func Middleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + seed, err := newSeed() + if err != nil { + http.Error(w, "failed to generate request seed", http.StatusInternalServerError) + return + } + next.ServeHTTP(w, r.WithContext(NewContext(r.Context(), seed))) + }) +} diff --git a/pkg/random/random_test.go b/pkg/random/random_test.go new file mode 100644 index 000000000..afbfe3d1d --- /dev/null +++ b/pkg/random/random_test.go @@ -0,0 +1,52 @@ +package random + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestGeneratorIsDeterministicForTheSameSeedAndLabel(t *testing.T) { + generator1, err := Generator(42, "label") + require.NoError(t, err) + generator2, err := Generator(42, "label") + require.NoError(t, err) + + var buf1, buf2 [16]byte + _, err = generator1.Read(buf1[:]) + require.NoError(t, err) + _, err = generator2.Read(buf2[:]) + require.NoError(t, err) + + require.Equal(t, buf1, buf2) +} + +func TestGeneratorDiffersForDifferentSeeds(t *testing.T) { + generator1, err := Generator(42, "label") + require.NoError(t, err) + generator2, err := Generator(43, "label") + require.NoError(t, err) + + var buf1, buf2 [16]byte + _, err = generator1.Read(buf1[:]) + require.NoError(t, err) + _, err = generator2.Read(buf2[:]) + require.NoError(t, err) + + require.NotEqual(t, buf1, buf2) +} + +func TestGeneratorDiffersForDifferentLabels(t *testing.T) { + generator1, err := Generator(42, "label-a") + require.NoError(t, err) + generator2, err := Generator(42, "label-b") + require.NoError(t, err) + + var buf1, buf2 [16]byte + _, err = generator1.Read(buf1[:]) + require.NoError(t, err) + _, err = generator2.Read(buf2[:]) + require.NoError(t, err) + + require.NotEqual(t, buf1, buf2) +} diff --git a/pkg/scd/models/models.go b/pkg/scd/models/models.go index 6dd839903..f4a456b1c 100644 --- a/pkg/scd/models/models.go +++ b/pkg/scd/models/models.go @@ -4,6 +4,7 @@ import ( "crypto/sha256" "encoding/base64" "fmt" + "io" "net/url" "strings" "time" @@ -38,6 +39,16 @@ type ( VersionNumber int32 ) +// NewDeterministicImplicitSubscriptionID derives a deterministic UUID from the given source. +// This allows the same ID to be generated across multiple Raft nodes and replays. +func NewDeterministicImplicitSubscriptionID(reader io.Reader) (dssmodels.ID, error) { + id, err := uuid.NewRandomFromReader(reader) + if err != nil { + return "", stacktrace.Propagate(err, "Failed to build UUID") + } + return dssmodels.ID(id.String()), nil +} + // NewOVNFromTime encodes t as an OVN. func NewOVNFromTime(t time.Time, salt string) OVN { sum := sha256.Sum256([]byte(salt + t.Format(time.RFC3339Nano))) diff --git a/pkg/scd/models/models_test.go b/pkg/scd/models/models_test.go index ee9d408e9..87e9590ce 100644 --- a/pkg/scd/models/models_test.go +++ b/pkg/scd/models/models_test.go @@ -1,6 +1,7 @@ package models import ( + "math/rand" "testing" "time" @@ -13,6 +14,36 @@ func TestOVNFromTimeIsValid(t *testing.T) { require.True(t, NewOVNFromTime(time.Now(), uuid.New().String()).Valid()) } +func TestNewDeterministicImplicitSubscriptionID(t *testing.T) { + t.Run("is a valid v4 UUID", func(t *testing.T) { + id, err := NewDeterministicImplicitSubscriptionID(rand.New(rand.NewSource(42))) + require.NoError(t, err) + _, err = dssmodels.IDFromString(id.String()) + require.NoError(t, err) + + parsed, err := uuid.Parse(id.String()) + require.NoError(t, err) + require.Equal(t, uuid.Version(4), parsed.Version()) + require.Equal(t, uuid.RFC4122, parsed.Variant()) + }) + + t.Run("is deterministic for identically-seeded sources", func(t *testing.T) { + id1, err := NewDeterministicImplicitSubscriptionID(rand.New(rand.NewSource(42))) + require.NoError(t, err) + id2, err := NewDeterministicImplicitSubscriptionID(rand.New(rand.NewSource(42))) + require.NoError(t, err) + require.Equal(t, id1, id2) + }) + + t.Run("differs when the source seed differs", func(t *testing.T) { + id1, err := NewDeterministicImplicitSubscriptionID(rand.New(rand.NewSource(42))) + require.NoError(t, err) + id2, err := NewDeterministicImplicitSubscriptionID(rand.New(rand.NewSource(43))) + require.NoError(t, err) + require.NotEqual(t, id1, id2) + }) +} + func TestNewOVNFromUUIDv7Suffix(t *testing.T) { type cases []struct { name string diff --git a/pkg/scd/operational_intents_handler.go b/pkg/scd/operational_intents_handler.go index ee2ec5d78..74df2bc04 100644 --- a/pkg/scd/operational_intents_handler.go +++ b/pkg/scd/operational_intents_handler.go @@ -5,12 +5,12 @@ import ( "time" "github.com/golang/geo/s2" - "github.com/google/uuid" "github.com/interuss/dss/pkg/api" restapi "github.com/interuss/dss/pkg/api/scdv1" "github.com/interuss/dss/pkg/auth" dsserr "github.com/interuss/dss/pkg/errors" dssmodels "github.com/interuss/dss/pkg/models" + "github.com/interuss/dss/pkg/random" "github.com/interuss/dss/pkg/scd/actions" scdmodels "github.com/interuss/dss/pkg/scd/models" "github.com/interuss/dss/pkg/scd/repos" @@ -403,8 +403,17 @@ func validateUpsertRequestAgainstPreviousOIR( // createAndStoreNewImplicitSubscription will create a brand new implicit subscription based on the provided parameters, // store it and return it. func createAndStoreNewImplicitSubscription(ctx context.Context, r repos.Repository, manager dssmodels.Manager, validParams *validOIRParams) (*scdmodels.Subscription, error) { + generator, err := random.Generator(random.MustFromContext(ctx), "implicit-subscription:"+validParams.id.String()) + if err != nil { + return nil, stacktrace.Propagate(err, "Failed to derive implicit subscription ID generator") + } + id, err := scdmodels.NewDeterministicImplicitSubscriptionID(generator) + if err != nil { + return nil, stacktrace.Propagate(err, "Failed to create implicit subscription ID") + } + subToUpsert := scdmodels.Subscription{ - ID: dssmodels.ID(uuid.New().String()), + ID: id, Manager: manager, StartTime: validParams.uExtent.StartTime, EndTime: validParams.uExtent.EndTime,