[raft/scd] Derive implicit subscription ID deterministically

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 {

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,

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}
 		}

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)))
+	})
+}

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)
+}

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)))

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

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,