[raft/memstore] Add Checkpoint/Restore

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/go-jose/go-jose/v4 v4.1.4
 	github.com/golang-jwt/jwt/v4 v4.5.2
 	github.com/golang/geo v0.0.0-20230421003525-6adc56603217
+	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/interuss/stacktrace v1.0.0
 	github.com/jackc/pgx/v5 v5.9.2

pkg/aux_/store/memstore/dss.go

@@ -2,24 +2,96 @@ package memstore
 
 import (
 	"context"
+	"database/sql"
 
 	auxmodels "github.com/interuss/dss/pkg/aux_/models"
 	dsserr "github.com/interuss/dss/pkg/errors"
+	"github.com/interuss/dss/pkg/timestamp"
 	"github.com/interuss/stacktrace"
 )
 
-func (r *repo) SaveOwnMetadata(_ context.Context, locality string, publicEndpoint string) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SaveOwnMetadata not implemented for memstore")
+func (r *repo) SaveOwnMetadata(ctx context.Context, locality string, publicEndpoint string) error {
+	if locality == "" {
+		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Locality not set")
+	}
+	if publicEndpoint == "" {
+		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Public endpoint not set")
+	}
+
+	now, err := timestamp.RequestTimestampFromContext(ctx)
+
+	if err != nil {
+		return err
+	}
+
+	r.state.Participants[locality] = &participant{
+		PublicEndpoint: publicEndpoint,
+		UpdatedAt:      now,
+	}
+	return nil
 }
 
 func (r *repo) GetDSSMetadata(_ context.Context) ([]*auxmodels.DSSMetadata, error) {
-	return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSMetadata not implemented for memstore")
+	metadata := make([]*auxmodels.DSSMetadata, 0, len(r.state.Participants))
+	for locality, p := range r.state.Participants {
+		updatedAt := p.UpdatedAt
+		m := &auxmodels.DSSMetadata{
+			Locality:       locality,
+			PublicEndpoint: p.PublicEndpoint,
+			UpdatedAt:      &updatedAt,
+		}
+
+		// Find the latest heartbeat across all sources for this locality.
+		var latest auxmodels.Heartbeat
+		found := false
+		for key, hb := range r.state.Heartbeats {
+			if key.Locality != locality {
+				continue
+			}
+			if !found || hb.Timestamp.After(*latest.Timestamp) {
+				latest = hb
+				found = true
+			}
+		}
+
+		if found {
+			m.LatestTimestamp.Source = sql.NullString{String: latest.Source, Valid: true}
+			m.LatestTimestamp.Timestamp = latest.Timestamp
+			m.LatestTimestamp.NextHeartbeatExpectedBefore = latest.NextHeartbeatExpectedBefore
+			m.LatestTimestamp.Reporter = sql.NullString{String: latest.Reporter, Valid: true}
+		}
+
+		metadata = append(metadata, m)
+	}
+	return metadata, nil
 }
 
-func (r *repo) RecordHeartbeat(_ context.Context, heartbeat auxmodels.Heartbeat) error {
-	return stacktrace.NewErrorWithCode(dsserr.NotImplemented, "RecordHeartbeat not implemented for memstore")
+func (r *repo) RecordHeartbeat(ctx context.Context, heartbeat auxmodels.Heartbeat) error {
+	if heartbeat.Locality == "" {
+		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Locality not set")
+	}
+	if heartbeat.Source == "" {
+		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Source not set")
+	}
+
+	if heartbeat.Timestamp == nil {
+
+		now, err := timestamp.RequestTimestampFromContext(ctx)
+
+		if err != nil {
+			return err
+		}
+		heartbeat.Timestamp = &now
+	}
+
+	if heartbeat.NextHeartbeatExpectedBefore != nil && heartbeat.NextHeartbeatExpectedBefore.Before(*heartbeat.Timestamp) {
+		return stacktrace.NewErrorWithCode(dsserr.BadRequest, "Cannot expect the timestamp of the next heartbeat before the timestamp of the new heartbeat")
+	}
+
+	r.state.Heartbeats[heartbeatKey{Locality: heartbeat.Locality, Source: heartbeat.Source}] = heartbeat
+	return nil
 }
 
 func (r *repo) GetDSSAirspaceRepresentationID(_ context.Context) (string, error) {
-	return "", stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSAirspaceRepresentationID not implemented for memstore")
+	return "", stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetDSSAirspaceRepresentationID not implementable for memstore")
 }

pkg/aux_/store/memstore/dss_test.go

@@ -0,0 +1,136 @@
+package memstore
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	auxmodels "github.com/interuss/dss/pkg/aux_/models"
+	dsserr "github.com/interuss/dss/pkg/errors"
+	"github.com/interuss/dss/pkg/timestamp"
+	"github.com/interuss/stacktrace"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+)
+
+var fakeClock = clockwork.NewFakeClock()
+
+func TestSaveOwnMetadataValidation(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.SaveOwnMetadata(ctx, "", "https://example.com")))
+	require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.SaveOwnMetadata(ctx, "dss-1", "")))
+}
+
+func TestSaveOwnMetadataRoundTrip(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
+
+	md, err := r.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+
+	require.Len(t, md, 1)
+	require.Equal(t, "dss-1", md[0].Locality)
+	require.Equal(t, "https://example.com", md[0].PublicEndpoint)
+	require.NotNil(t, md[0].UpdatedAt)
+
+	// No heartbeat recorded yet.
+	require.False(t, md[0].LatestTimestamp.Source.Valid)
+	require.Nil(t, md[0].LatestTimestamp.Timestamp)
+}
+
+func TestSaveOwnMetadataUpsert(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://old.example.com"))
+	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://new.example.com"))
+
+	md, err := r.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+
+	require.Len(t, md, 1)
+	require.Equal(t, "https://new.example.com", md[0].PublicEndpoint)
+}
+
+func TestRecordHeartbeatValidation(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Source: "source1"})))
+	require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1"})))
+
+	ts := time.Now()
+	before := ts.Add(-time.Minute)
+	err := r.RecordHeartbeat(ctx, auxmodels.Heartbeat{
+		Locality:                    "dss-1",
+		Source:                      "source1",
+		Timestamp:                   &ts,
+		NextHeartbeatExpectedBefore: &before,
+	})
+
+	require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(err))
+}
+
+func TestRecordHeartbeatDefaultsTimestamp(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
+	require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1"}))
+
+	md, err := r.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+
+	require.Len(t, md, 1)
+	require.True(t, md[0].LatestTimestamp.Source.Valid)
+	require.NotNil(t, md[0].LatestTimestamp.Timestamp)
+}
+
+func TestGetDSSMetadataPicksLatestHeartbeat(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
+
+	older := time.Now().Add(-time.Hour)
+	newer := time.Now()
+	require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1", Timestamp: &older, Reporter: "uss1"}))
+	require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source2", Timestamp: &newer, Reporter: "uss2"}))
+
+	md, err := r.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+
+	require.Len(t, md, 1)
+	require.True(t, md[0].LatestTimestamp.Timestamp.Equal(newer))
+	require.Equal(t, "source2", md[0].LatestTimestamp.Source.String)
+	require.Equal(t, "uss2", md[0].LatestTimestamp.Reporter.String)
+}
+
+func TestGetDSSMetadataUpdatesHeartbeatPerSource(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	r := newRepo()
+
+	require.NoError(t, r.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
+
+	first := time.Now().Add(-time.Hour)
+	second := time.Now()
+	require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1", Timestamp: &first}))
+	require.NoError(t, r.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source1", Timestamp: &second}))
+
+	md, err := r.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+
+	require.Len(t, md, 1)
+	require.True(t, md[0].LatestTimestamp.Timestamp.Equal(second))
+}

pkg/aux_/store/memstore/snapshot.go

@@ -0,0 +1,35 @@
+package memstore
+
+import (
+	"bytes"
+	"encoding/gob"
+
+	"github.com/interuss/stacktrace"
+)
+
+const snapshotVersion = 1
+
+type snapshotEnvelope struct {
+	Version int
+	State   state
+}
+
+func (r *repo) GetSnapshot() ([]byte, error) {
+	var buf bytes.Buffer
+	if err := gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion, State: r.state}); err != nil {
+		return nil, stacktrace.Propagate(err, "Failed to encode memstore snapshot")
+	}
+	return buf.Bytes(), nil
+}
+
+func (r *repo) RestoreFromSnapshot(data []byte) error {
+	var env snapshotEnvelope
+	if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&env); err != nil {
+		return stacktrace.Propagate(err, "Failed to decode memstore snapshot")
+	}
+	if env.Version != snapshotVersion {
+		return stacktrace.NewError("Unsupported memstore snapshot version %d, expected %d", env.Version, snapshotVersion)
+	}
+	r.state = env.State
+	return nil
+}

pkg/aux_/store/memstore/snapshot_test.go

@@ -0,0 +1,67 @@
+package memstore
+
+import (
+	"bytes"
+	"context"
+	"encoding/gob"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	auxmodels "github.com/interuss/dss/pkg/aux_/models"
+	"github.com/interuss/dss/pkg/models"
+	"github.com/interuss/dss/pkg/timestamp"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSnapshotRoundTrip(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	src := newRepo()
+	require.NoError(t, src.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
+	ts := time.Now().UTC()
+	require.NoError(t, src.RecordHeartbeat(ctx, auxmodels.Heartbeat{Locality: "dss-1", Source: "source-1", Timestamp: &ts, Reporter: "uss-1"}))
+
+	data, err := src.GetSnapshot()
+	require.NoError(t, err)
+
+	dst := newRepo()
+	require.NoError(t, dst.RestoreFromSnapshot(data))
+
+	want, err := src.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+	got, err := dst.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+	if diff := cmp.Diff(want, got, cmpopts.EquateApproxTime(0), cmp.AllowUnexported(models.Version{})); diff != "" {
+		t.Errorf("Store mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func TestRestoreFromSnapshotReplacesState(t *testing.T) {
+	ctx := context.Background()
+	ctx = timestamp.WithRequestTimestamp(ctx, fakeClock.Now())
+	src := newRepo()
+	require.NoError(t, src.SaveOwnMetadata(ctx, "dss-1", "https://example.com"))
+	data, err := src.GetSnapshot()
+	require.NoError(t, err)
+
+	dst := newRepo()
+	require.NoError(t, dst.SaveOwnMetadata(ctx, "dss-2", "https://other.example.com"))
+	require.NoError(t, dst.RestoreFromSnapshot(data))
+
+	md, err := dst.GetDSSMetadata(ctx)
+	require.NoError(t, err)
+	require.Len(t, md, 1)
+	require.Equal(t, "dss-1", md[0].Locality)
+}
+
+func TestRestoreFromSnapshotInvalidData(t *testing.T) {
+	require.Error(t, newRepo().RestoreFromSnapshot([]byte("random value that is definitely not valid")))
+}
+
+func TestRestoreFromSnapshotVersionMismatch(t *testing.T) {
+	var buf bytes.Buffer
+	require.NoError(t, gob.NewEncoder(&buf).Encode(snapshotEnvelope{Version: snapshotVersion + 1}))
+	require.Error(t, newRepo().RestoreFromSnapshot(buf.Bytes()))
+}