[raft] Make raft runnable-locally

Makefile

@@ -157,7 +157,6 @@ test-go-units-crdb: cleanup-test-go-units-crdb
 	go run ./cmds/db-manager/main.go migrate --schemas_dir ./build/db_schemas/scd --db_version latest --datastore_host localhost
 	go run ./cmds/db-manager/main.go migrate --schemas_dir ./build/db_schemas/aux_ --db_version latest --datastore_host localhost
 	go test -cover -count=1 -v ./pkg/rid/store/sqlstore --datastore_host localhost --datastore_port 26257 --datastore_ssl_mode disable --datastore_user root -test.gocoverdir=$(COVERDATA_DIR)
-	go test -cover -count=1 -v ./pkg/rid/application --datastore_host localhost --datastore_port 26257 --datastore_ssl_mode disable --datastore_user root -test.gocoverdir=$(COVERDATA_DIR)
 	go test -cover -count=1 -v ./pkg/scd/store/sqlstore --datastore_host localhost --datastore_port 26257 --datastore_ssl_mode disable --datastore_user root -test.gocoverdir=$(COVERDATA_DIR)
 	go test -cover -count=1 -v ./pkg/aux_/store/sqlstore --datastore_host localhost --datastore_port 26257 --datastore_ssl_mode disable --datastore_user root -test.gocoverdir=$(COVERDATA_DIR)
 	@docker stop dss-crdb-for-testing > /dev/null

build/dev/docker-compose_dss.yaml

@@ -111,6 +111,7 @@ services:
       - $PWD/../test-certs:/var/test-certs:ro
       - $PWD/startup/core_service.sh:/startup/core_service.sh:ro
       - $PWD/startup/coverdata:/startup/coverdata:rw  # we will save coverage info here
+      - raftdata:/raftdata
     environment:
       COMPOSE_PROFILES: ${COMPOSE_PROFILES}
       # Note: requires the Dockerfile to have been built with "-cover" in the EXTRA_GO_INSTALL_FLAGS var
@@ -147,7 +148,7 @@ services:
       interval: 3m
       start_period: 30s
       start_interval: 5s
-    profiles: ["", "with-yugabyte", "with-monitoring"]
+    profiles: ["", "with-yugabyte", "with-monitoring", "with-raft"]
 
   local-dss-dummy-oauth:
     build:
@@ -216,3 +217,4 @@ networks:
 
 volumes:
   local-dss-data:
+  raftdata:

build/dev/startup/core_service.sh

@@ -10,8 +10,13 @@ if [ "${COMPOSE_PROFILES#*"with-yugabyte"}" != "${COMPOSE_PROFILES}" ]; then
   echo "Using Yugabyte"
   DATASTORE_CONNECTION="-datastore_host local-dss-ybdb -datastore_user yugabyte --datastore_port 5433"
 else
-  echo "Using CockroachDB"
-  DATASTORE_CONNECTION="-datastore_host local-dss-crdb"
+    if [ "${COMPOSE_PROFILES#*"with-raft"}" != "${COMPOSE_PROFILES}" ]; then
+      echo "Using raft"
+      DATASTORE_CONNECTION="-store_type raft -raft_node_id=1 -rid_raft_peers=1=http://127.0.0.1:9011 -scd_raft_peers=1=http://127.0.0.1:9021 -aux_raft_peers=1=http://127.0.0.1:9031 -raft_datadir /raftdata"
+    else
+      echo "Using CockroachDB"
+      DATASTORE_CONNECTION="-datastore_host local-dss-crdb"
+    fi
 fi
 
 if [ "${COMPOSE_PROFILES#*"with-monitoring"}" != "${COMPOSE_PROFILES}" ]; then

cmds/core-service/main.go

@@ -27,7 +27,6 @@ import (
 	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"
 	rids "github.com/interuss/dss/pkg/rid/store"
@@ -112,6 +111,15 @@ func createAuxServer(ctx context.Context, locality string, publicEndpoint string
 		return nil, stacktrace.Propagate(err, "Unable to interact with store")
 	}
 
+	ctx = timestamp.NewContext(ctx, time.Now())
+
+	seed, err := random.NewSeed()
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "Unable to generate seed")
+	}
+
+	ctx = random.NewContext(ctx, seed)
+
 	err = repo.SaveOwnMetadata(ctx, locality, publicEndpoint)
 
 	if err != nil {
@@ -141,15 +149,12 @@ func createRIDServers(ctx context.Context, locality string, logger *zap.Logger)
 		}
 	}
 
-	app := application.NewFromTransactor(ridStore, logger)
 	return &rid_v1.Server{
 			Store:             ridStore,
-			App:               app,
 			Locality:          locality,
 			AllowHTTPBaseUrls: *allowHTTPBaseUrls,
 		}, &rid_v2.Server{
 			Store:             ridStore,
-			App:               app,
 			Locality:          locality,
 			AllowHTTPBaseUrls: *allowHTTPBaseUrls,
 		}, nil
@@ -368,9 +373,9 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st
 	handler = authorizer.TokenMiddleware(handler)
 	handler = http.TimeoutHandler(handler, *timeout, "request timeout")
 	handler = logging.HTTPMiddleware(logger, *dumpRequests, handler)
-	handler = timestamp.RequestTimestampMiddleware(handler)
+	handler = timestamp.Middleware(handler)
 	handler = random.Middleware(handler)
-	handler = requestlocality.LocalityMiddleware(locality)(handler)
+	handler = requestlocality.Middleware(locality)(handler)
 
 	if *enableMetrics || *enableTracing {
 		// We use the default settings; the APIRouter handler will override the span value accordingly, as it has more information.

pkg/aux_/pool_participants.go

@@ -96,7 +96,7 @@ func (a *Server) PutDSSInstancesHeartbeat(ctx context.Context, req *restapi.PutD
 		}
 		heartbeat.Timestamp = &ts
 	} else {
-		now := timestamp.MustGetRequestTimestamp(ctx)
+		now := timestamp.MustFromContext(ctx)
 		heartbeat.Timestamp = &now
 	}
 

pkg/aux_/store/memstore/dss.go

@@ -11,7 +11,7 @@ import (
 )
 
 func (r *repo) SaveOwnMetadata(ctx context.Context, loc string, publicEndpoint string) error {
-	now := timestamp.MustGetRequestTimestamp(ctx)
+	now := timestamp.MustFromContext(ctx)
 
 	r.state.Participants[locality(loc)] = &participant{
 		PublicEndpoint: publicEndpoint,

pkg/aux_/store/memstore/dss_test.go

@@ -15,7 +15,7 @@ var fakeClock = clockwork.NewFakeClock()
 
 func TestSaveOwnMetadataRoundTrip(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	r := newRepo()
 
 	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
@@ -35,7 +35,7 @@ func TestSaveOwnMetadataRoundTrip(t *testing.T) {
 
 func TestSaveOwnMetadataUpsert(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	r := newRepo()
 
 	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://old.example.com"))
@@ -50,7 +50,7 @@ func TestSaveOwnMetadataUpsert(t *testing.T) {
 
 func TestGetDSSMetadataPicksLatestHeartbeat(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	r := newRepo()
 
 	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
@@ -71,7 +71,7 @@ func TestGetDSSMetadataPicksLatestHeartbeat(t *testing.T) {
 
 func TestGetDSSMetadataUpdatesHeartbeatPerSource(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	r := newRepo()
 
 	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))

pkg/aux_/store/memstore/snapshot_test.go

@@ -17,7 +17,7 @@ import (
 
 func TestSnapshotRoundTrip(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	src := newRepo()
 	require.NoError(t, src.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
 	ts := time.Now().UTC()
@@ -40,7 +40,7 @@ func TestSnapshotRoundTrip(t *testing.T) {
 
 func TestRestoreFromSnapshotReplacesState(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	src := newRepo()
 	require.NoError(t, src.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
 	data, err := src.GetSnapshot()

pkg/aux_/store/memstore/store_test.go

@@ -10,7 +10,7 @@ import (
 
 func TestCheckpointRestore(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 
 	r := newRepo()
 
@@ -34,7 +34,7 @@ func TestCheckpointRestore(t *testing.T) {
 
 func TestCheckpointIsolatesUpsert(t *testing.T) {
 	ctx := context.Background()
-	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	ctx = timestamp.NewContext(ctx, fakeClock.Now())
 	r := newRepo()
 
 	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://old.example.com"))

pkg/aux_/store/raftstore/store.go

@@ -24,8 +24,7 @@ const (
 // repo is a full implementation of aux_.repos.Repository for Raft-based storage.
 type repo struct {
 	consensus *consensus.Consensus
-	memStore  *memstore.Store[repos.Repository]
-	memRepo   repos.Repository
+	*memstore.Store[repos.Repository]
 }
 
 func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.Store[repos.Repository], error) {
@@ -39,7 +38,7 @@ func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.
 		return nil, stacktrace.Propagate(err, "failed to initialize aux memstore")
 	}
 
-	r := &repo{memStore: memStore, memRepo: memStore.GetRepo()}
+	r := &repo{Store: memStore}
 	store, err := raftstore.Init(ctx, logger.With(zap.String("service", "aux_")), locality, params, r, nil)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to initialize aux raftstore")
@@ -52,14 +51,6 @@ func Init(ctx context.Context, logger *zap.Logger, locality string) (*raftstore.
 
 func (r *repo) GetRepo() repos.Repository { return r }
 
-func (r *repo) GetSnapshot() ([]byte, error) {
-	return r.memStore.GetSnapshot()
-}
-
-func (r *repo) RestoreFromSnapshot(data []byte) error {
-	return r.memStore.RestoreFromSnapshot(data)
-}
-
 func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, error) {
 	switch proposal.RequestType {
 	case saveOwnMetadata:
@@ -68,18 +59,18 @@ func (r *repo) Apply(ctx context.Context, proposal consensus.Proposal) (any, err
 			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", saveOwnMetadata)
 		}
 
-		return nil, r.memRepo.SaveOwnMetadata(ctx, payload.Locality, payload.PublicEndpoint)
+		return nil, r.Store.GetRepo().SaveOwnMetadata(ctx, payload.Locality, payload.PublicEndpoint)
 
 	case getDSSMetadata:
-		return r.memRepo.GetDSSMetadata(ctx)
+		return r.Store.GetRepo().GetDSSMetadata(ctx)
 
 	case recordHeartbeat:
 		var heartbeat auxmodels.Heartbeat
 		if err := json.Unmarshal(proposal.Value, &heartbeat); err != nil {
 			return nil, stacktrace.Propagate(err, "failed to unmarshal %s payload", recordHeartbeat)
 		}
 
-		return nil, r.memRepo.RecordHeartbeat(ctx, heartbeat)
+		return nil, r.Store.GetRepo().RecordHeartbeat(ctx, heartbeat)
 
 	default:
 		return nil, stacktrace.NewError("unknown request type: %q", proposal.RequestType)

pkg/errors/errors.go

@@ -16,10 +16,6 @@ const (
 	// larger than the max area allowed. See geo/s2.go.
 	AreaTooLarge = stacktrace.ErrorCode(iota)
 
-	// MissingOVNs is the error to signal that an AirspaceConflictResponse should
-	// be returned rather than the standard error response.
-	MissingOVNs
-
 	// AlreadyExists is used when attempting to create a resource that already
 	// exists.
 	AlreadyExists

pkg/locality/locality.go

@@ -7,30 +7,30 @@ import (
 	"github.com/interuss/stacktrace"
 )
 
-type localityKey struct{}
+type key struct{}
 
-// MustGetRequestLocality returns the request locality from the context and panics if it is not
+// MustFromContext returns the request locality from the context and panics if it is not
 // present, which is a programming error.
-func MustGetRequestLocality(ctx context.Context) string {
-	locality, ok := ctx.Value(localityKey{}).(string)
+func MustFromContext(ctx context.Context) string {
+	locality, ok := ctx.Value(key{}).(string)
 	if !ok {
 		panic(stacktrace.NewError("request locality not present in context"))
 	}
 
 	return locality
 }
 
-// WithRequestLocality returns a new context with the given locality.
-func WithRequestLocality(ctx context.Context, locality string) context.Context {
-	return context.WithValue(ctx, localityKey{}, locality)
+// NewContext returns a new context with the given locality.
+func NewContext(ctx context.Context, locality string) context.Context {
+	return context.WithValue(ctx, key{}, locality)
 }
 
-// LocalityMiddleware is an HTTP middleware that stamps each incoming request with this
+// Middleware is an HTTP middleware that stamps each incoming request with this
 // DSS instance's locality so that locality-dependent operations execute deterministically across nodes.
-func LocalityMiddleware(locality string) func(http.Handler) http.Handler {
+func Middleware(locality string) func(http.Handler) http.Handler {
 	return func(next http.Handler) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			next.ServeHTTP(w, r.WithContext(WithRequestLocality(r.Context(), locality)))
+			next.ServeHTTP(w, r.WithContext(NewContext(r.Context(), locality)))
 		})
 	}
 }

pkg/models/geo.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"encoding/json"
 	"time"
 
 	"github.com/golang/geo/s2"
@@ -46,6 +47,83 @@ type Volume3D struct {
 	Footprint Geometry
 }
 
+type Volume3DJSON struct {
+	AltitudeHi *float32      `json:"altitude_hi,omitempty"`
+	AltitudeLo *float32      `json:"altitude_lo,omitempty"`
+	Footprint  *geometryJSON `json:"footprint,omitempty"`
+}
+
+type geometryType string
+
+const (
+	circle  geometryType = "circle"
+	polygon geometryType = "polygon"
+	cells   geometryType = "cells"
+)
+
+// geometryJSON is a helper struct for marshaling and unmarshaling Geometry types to/from JSON.
+type geometryJSON struct {
+	Type    geometryType `json:"type"`
+	Polygon *GeoPolygon  `json:"polygon,omitempty"`
+	Circle  *GeoCircle   `json:"circle,omitempty"`
+	Cells   []s2.CellID  `json:"cells,omitempty"`
+}
+
+func (v Volume3D) MarshalJSON() ([]byte, error) {
+	w := Volume3DJSON{AltitudeHi: v.AltitudeHi, AltitudeLo: v.AltitudeLo}
+	if v.Footprint != nil {
+		switch f := v.Footprint.(type) {
+		case *GeoPolygon:
+			w.Footprint = &geometryJSON{Type: polygon, Polygon: f}
+
+		case *GeoCircle:
+			w.Footprint = &geometryJSON{Type: circle, Circle: f}
+
+		case precomputedCellGeometry:
+			cellsResult := make([]s2.CellID, 0, len(f))
+			for id := range f {
+				cellsResult = append(cellsResult, id)
+			}
+			w.Footprint = &geometryJSON{Type: cells, Cells: cellsResult}
+
+		default:
+			return nil, stacktrace.NewError("Volume3D: unsupported Footprint type %T for JSON marshaling", v.Footprint)
+		}
+	}
+
+	return json.Marshal(w)
+}
+
+func (v *Volume3D) UnmarshalJSON(data []byte) error {
+	var w Volume3DJSON
+	if err := json.Unmarshal(data, &w); err != nil {
+		return err
+	}
+	v.AltitudeHi = w.AltitudeHi
+	v.AltitudeLo = w.AltitudeLo
+	if w.Footprint != nil {
+		switch w.Footprint.Type {
+		case polygon:
+			v.Footprint = w.Footprint.Polygon
+
+		case circle:
+			v.Footprint = w.Footprint.Circle
+
+		case cells:
+			pcg := make(precomputedCellGeometry, len(w.Footprint.Cells))
+			for _, id := range w.Footprint.Cells {
+				pcg[id] = struct{}{}
+			}
+
+			v.Footprint = pcg
+		default:
+			return stacktrace.NewError("Volume3D: unknown geometry type %q", w.Footprint.Type)
+		}
+	}
+
+	return nil
+}
+
 // Geometry models a geometry.
 type Geometry interface {
 	// CalculateCovering returns an s2 cell covering for a geometry.