feat(store): add WaitForStoreVersion API to wait for cluster store replication up to target version

internal/cli/cli.go

@@ -458,7 +458,7 @@ func (cli *CLI) AddMachine(ctx context.Context, opts AddMachineOptions) (_ *clie
 	}
 
 	// Snapshot the cluster store version so the new machine can catch up before participating.
-	var storeVersion map[string]int64
+	var storeVersion map[string]uint64
 	inspectResp, err = c.MachineClient.InspectMachine(ctx, &emptypb.Empty{})
 	if err != nil {
 		// TODO(lhf): remove Unimplemented check when v0.17.0 is released.

internal/machine/cluster.go

@@ -446,11 +446,11 @@ func (cc *clusterController) waitStoreSync(ctx context.Context) error {
 }
 
 // laggingActors returns target actors whose local version is below the required value, as [have, need].
-func laggingActors(local, target map[string]int64) map[string][2]int64 {
-	lagging := make(map[string][2]int64)
+func laggingActors(local, target map[string]uint64) map[string][2]uint64 {
+	lagging := make(map[string][2]uint64)
 	for actor, need := range target {
 		if have := local[actor]; have < need {
-			lagging[actor] = [2]int64{have, need}
+			lagging[actor] = [2]uint64{have, need}
 		}
 	}
 	return lagging

internal/machine/machine.go

@@ -1124,6 +1124,19 @@ func (m *Machine) InspectMachine(ctx context.Context, _ *emptypb.Empty) (*pb.Ins
 	}, nil
 }
 
+func (m *Machine) WaitForStoreVersion(ctx context.Context, req *pb.WaitForStoreVersionRequest) (*emptypb.Empty, error) {
+	if err := m.store.WaitForVersion(ctx, req.MinVersion); err != nil {
+		if ctx.Err() != nil {
+			return nil, status.FromContextError(ctx.Err()).Err()
+		}
+		if errors.Is(err, store.ErrInvalidStoreVersion) {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	return &emptypb.Empty{}, nil
+}
+
 // UpdateMachine updates the configuration of this machine in its local state (the source of truth) and syncs
 // the result to the cluster store.
 func (m *Machine) UpdateMachine(ctx context.Context, req *pb.UpdateMachineRequest) (*pb.UpdateMachineResponse, error) {

internal/machine/state.go

@@ -29,9 +29,9 @@ type State struct {
 	// PublicIP is this machine's advertised public IP for ingress. Zero value if unset.
 	PublicIP netip.Addr `json:",omitempty"`
 	// MinStoreVersion is the cluster store version this machine must reach before participating.
-	// Per-actor vector (Corrosion actor UUID → max applied db_version) captured from an existing
+	// Per-actor vector (Corrosion actor UUID → max processed db_version) captured from an existing
 	// member at join time. Cleared once reached.
-	MinStoreVersion map[string]int64 `json:",omitempty"`
+	MinStoreVersion map[string]uint64 `json:",omitempty"`
 	// CorrosionAPIToken authenticates requests to the local Corrosion API.
 	CorrosionAPIToken secret.Secret `json:",omitempty"`
 

internal/machine/store/replication.go

@@ -0,0 +1,106 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+// ErrInvalidStoreVersion indicates an invalid actor UUID in a store version vector.
+var ErrInvalidStoreVersion = errors.New("invalid store version")
+
+// WaitForVersion waits until the local store has reached each actor's minimum version in minVersion, with no known
+// missing or pending transactions through those versions. Corrosion may satisfy a version by applying its surviving
+// changes or by marking it complete because its changes have been superseded.
+//
+// Waiting normally makes the captured data available locally. However, if another write replaces some of that data
+// before it arrives, Corrosion can complete the older version without transferring the replaced data. If the
+// replacement is outside minVersion, this method can return while the affected data is still missing or outdated.
+// This can happen during concurrent updates even when all machines are well connected.
+//
+// This method does not guarantee an exact snapshot or delivery of every historical value.
+// Callers that require a specific record or condition should verify it after waiting.
+//
+// The method observes native replication without initiating synchronisation.
+// An empty minVersion requires no replication. The context controls cancellation and the deadline.
+func (s *Store) WaitForVersion(ctx context.Context, minVersion map[string]uint64) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	placeholders := make([]string, 0, len(minVersion))
+	args := make([]any, 0, 2*len(minVersion))
+	for actor, version := range minVersion {
+		actorID, err := uuid.Parse(actor)
+		if err != nil {
+			return fmt.Errorf("%w: actor '%s' is not a UUID: %v", ErrInvalidStoreVersion, actor, err)
+		}
+		if version == 0 {
+			continue
+		}
+
+		// Corrosion accepts a JSON array of bytes as a BLOB parameter. A UUID marshals as a string,
+		// and a byte slice marshals as base64, so use a plain fixed-size array instead.
+		placeholders = append(placeholders, "(?, ?)")
+		args = append(args, [16]byte(actorID), version)
+	}
+	if len(args) == 0 {
+		return nil
+	}
+
+	// Check all bookkeeping in one SQLite snapshot. Raw buffered rows can remain after application,
+	// but sequence bookkeeping is removed in the same transaction that applies or clears a version.
+	query := `WITH min_version(actor_id, version) AS (VALUES ` + strings.Join(placeholders, ", ") + `)
+		SELECT NOT EXISTS (
+			SELECT 1 FROM min_version
+			LEFT JOIN crsql_db_versions AS current ON current.site_id = min_version.actor_id
+			WHERE COALESCE(current.db_version, 0) < min_version.version
+				OR EXISTS (
+					SELECT 1 FROM __corro_bookkeeping_gaps
+					WHERE actor_id = min_version.actor_id AND start <= min_version.version
+				)
+				OR EXISTS (
+					SELECT 1 FROM __corro_seq_bookkeeping
+					WHERE site_id = min_version.actor_id AND db_version <= min_version.version
+				)
+		)`
+
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		rows, err := s.corro.QueryContext(ctx, query, args...)
+		if err != nil {
+			return fmt.Errorf("check store replication: %w", err)
+		}
+		if !rows.Next() {
+			if err = rows.Err(); err != nil {
+				return fmt.Errorf("check store replication: %w", err)
+			}
+			return errors.New("check store replication: store version query returned no rows")
+		}
+		var reached int
+		if err = rows.Scan(&reached); err != nil {
+			rows.Close()
+			return fmt.Errorf("check store replication: %w", err)
+		}
+		rows.Next() // Consume the end-of-query event.
+		rows.Close()
+		if err = rows.Err(); err != nil {
+			return fmt.Errorf("check store replication: %w", err)
+		}
+		if reached == 1 {
+			return ctx.Err()
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}

internal/machine/store/store.go

@@ -73,20 +73,27 @@ func (s *Store) Delete(ctx context.Context, key string) error {
 	return err
 }
 
-// Version returns the cluster store's per-actor version vector:
-// Corrosion actor ID (UUID string) → max received db_version for that actor.
-func (s *Store) Version(ctx context.Context) (map[string]int64, error) {
+// Version returns the replication progress observed by this store.
+// Each entry maps a Corrosion actor UUID to its highest processed database version. Corrosion also counts versions
+// whose changes were superseded and skipped. Versions at or below a reported version may still be missing or pending
+// locally.
+//
+// You can use [Store.WaitForVersion] to wait for replication through these versions, subject to the data-availability
+// limitations documented there.
+//
+// Capturing a vector does not wait for replication or prevent further writes.
+func (s *Store) Version(ctx context.Context) (map[string]uint64, error) {
 	rows, err := s.corro.QueryContext(ctx, "SELECT site_id, db_version FROM crsql_db_versions")
 	if err != nil {
 		return nil, fmt.Errorf("query crsql_db_versions: %w", err)
 	}
 	defer rows.Close()
 
-	versions := make(map[string]int64)
+	versions := make(map[string]uint64)
 	for rows.Next() {
 		var (
 			siteID  []byte
-			version int64
+			version uint64
 		)
 		if err = rows.Scan(&siteID, &version); err != nil {
 			return nil, fmt.Errorf("scan actor version: %w", err)
@@ -102,8 +109,8 @@ func (s *Store) Version(ctx context.Context) (map[string]int64, error) {
 
 type MissingChange struct {
 	ActorID      string
-	StartVersion int64
-	EndVersion   int64
+	StartVersion uint64
+	EndVersion   uint64
 }
 
 // KnownMissingChanges returns a list of currently known missing changes in the Corrosion database.