restore to state on main

go.mod

@@ -23,13 +23,13 @@ require (
 	github.com/ncruces/go-sqlite3 v0.29.0
 	github.com/ncruces/zenity v0.10.14
 	github.com/prometheus/client_golang v1.22.0
-	github.com/stretchr/testify v1.10.0
 	github.com/vearutop/statigz v1.5.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.41.0
 	golang.org/x/net v0.43.0
 	golang.org/x/sync v0.17.0
+	golang.org/x/sys v0.36.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2
 	google.golang.org/grpc v1.72.0
 	google.golang.org/protobuf v1.36.6
@@ -40,7 +40,6 @@ require (
 	github.com/akavel/rsrc v0.10.2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dchest/jsmin v1.0.0 // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/getlantern/context v0.0.0-20220418194847-3d5e7a086201 // indirect
@@ -59,7 +58,6 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/ncruces/julianday v1.0.0 // indirect
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.63.0 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
@@ -70,15 +68,6 @@ require (
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
 	golang.org/x/image v0.27.0 // indirect
-	golang.org/x/sys v0.36.0 // indirect
 	golang.org/x/text v0.29.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 // indirect
-<<<<<<< HEAD
-=======
-	gopkg.in/yaml.v3 v3.0.1 // indirect
-	modernc.org/libc v1.65.0 // indirect
-	modernc.org/mathutil v1.7.1 // indirect
-	modernc.org/memory v1.10.0 // indirect
-	modernc.org/sqlite v1.37.0 // indirect
->>>>>>> 9041d3c (improve sync api security by using 'Authorization' headers for initial key exchange)
 )

go.sum

@@ -93,12 +93,8 @@ github.com/josephspurrier/goversioninfo v1.5.0/go.mod h1:6MoTvFZ6GKJkzcdLnU5T/RG
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
-github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
-github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/lxn/walk v0.0.0-20210112085537-c389da54e794/go.mod h1:E23UucZGqpuUANJooIbHWCufXvOcT6E7Stq81gU+CSQ=
@@ -136,13 +132,6 @@ github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzM
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
 github.com/randall77/makefat v0.0.0-20210315173500-7ddd0e42c844 h1:GranzK4hv1/pqTIhMTXt2X8MmMOuH3hMeUR0o9SP5yc=
 github.com/randall77/makefat v0.0.0-20210315173500-7ddd0e42c844/go.mod h1:T1TLSfyWVBRXVGzWd0o9BI4kfoO9InEgfQe4NV3mLz8=
-<<<<<<< HEAD
-=======
-github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
-github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
-github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
-github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
->>>>>>> 9041d3c (improve sync api security by using 'Authorization' headers for initial key exchange)
 github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
@@ -235,8 +224,6 @@ google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/
 gopkg.in/Knetic/govaluate.v3 v3.0.0/go.mod h1:csKLBORsPbafmSCGTEh3U7Ozmsuq8ZSIlKk1bcqph0E=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
-gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
 gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

internal/api/syncapi/authentication.go

@@ -1,192 +0,0 @@
-package syncapi
-
-import (
-	"context"
-	"encoding/base64"
-	"errors"
-	"fmt"
-	"net/http"
-	"slices"
-	"strings"
-
-	v1 "github.com/garethgeorge/backrest/gen/go/v1"
-	"github.com/garethgeorge/backrest/internal/config"
-	"github.com/garethgeorge/backrest/internal/cryptoutil"
-	"go.uber.org/zap"
-	"google.golang.org/protobuf/proto"
-)
-
-type peerContextKey struct{}
-type peerPublicKeyContextKey struct{}
-
-func PeerFromContext(ctx context.Context) *v1.Multihost_Peer {
-	return ctx.Value(peerContextKey{}).(*v1.Multihost_Peer)
-}
-
-func PeerPublicKeyFromContext(ctx context.Context) *cryptoutil.PublicKey {
-	return ctx.Value(peerPublicKeyContextKey{}).(*cryptoutil.PublicKey)
-}
-
-func ContextWithPeer(ctx context.Context, peer *v1.Multihost_Peer, publicKey *cryptoutil.PublicKey) context.Context {
-	if peer == nil {
-		return ctx
-	}
-	ctx = context.WithValue(ctx, peerContextKey{}, peer)
-	ctx = context.WithValue(ctx, peerPublicKeyContextKey{}, publicKey)
-	return ctx
-}
-
-// HTTP decorator for authentication middleware.
-func AuthenticationMiddleware(configManager *config.ConfigManager, handler http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		authHeader := r.Header.Get("Authorization")
-		if authHeader == "" {
-			http.Error(w, "Unauthorized: missing authentication header", http.StatusUnauthorized)
-			return
-		}
-
-		config, err := configManager.Get()
-		if err != nil {
-			http.Error(w, "Internal error", http.StatusInternalServerError)
-			return
-		}
-		authorizedClientPeers := config.GetMultihost().GetAuthorizedClients()
-
-		peerKey, instanceID, err := verifyAuthenticationHeader(authHeader)
-		if err != nil {
-			http.Error(w, fmt.Sprintf("Unauthorized: %v", err), http.StatusUnauthorized)
-			return
-		}
-
-		authorizedPeerIdx := slices.IndexFunc(authorizedClientPeers, func(peer *v1.Multihost_Peer) bool {
-			return peer.Keyid == peerKey.KeyID()
-		})
-		if authorizedPeerIdx == -1 {
-			zap.S().Errorf("peer key %q is not listed in authorized clients", peerKey.KeyID())
-			http.Error(w, fmt.Sprintf("Unauthorized: peer key %q is not listed in authorized clients", peerKey.KeyID()), http.StatusUnauthorized)
-			return
-		}
-		authorizedPeer := authorizedClientPeers[authorizedPeerIdx]
-		if authorizedPeer.InstanceId != instanceID {
-			zap.S().Errorf("instance ID mismatch for peer key %q, expected %q, got %q", peerKey.KeyID(), authorizedPeer.InstanceId, instanceID)
-			http.Error(w, fmt.Sprintf("Unauthorized: instance ID mismatch for peer key %q, expected %q, got %q", peerKey.KeyID(), authorizedPeer.InstanceId, instanceID), http.StatusUnauthorized)
-			return
-		}
-		ctx := ContextWithPeer(r.Context(), authorizedPeer, peerKey)
-		handler.ServeHTTP(w, r.WithContext(ctx))
-	})
-}
-
-func createHandshakePacket(instanceID string, identity *cryptoutil.PrivateKey) (*v1.SyncStreamItem, error) {
-	signedMessage, err := createSignedMessage([]byte(instanceID), identity)
-	if err != nil {
-		return nil, fmt.Errorf("signing instance ID: %w", err)
-	}
-
-	return &v1.SyncStreamItem{
-		Action: &v1.SyncStreamItem_Handshake{
-			Handshake: &v1.SyncStreamItem_SyncActionHandshake{
-				ProtocolVersion: SyncProtocolVersion,
-				InstanceId:      signedMessage,
-				PublicKey:       identity.PublicKeyProto(),
-			},
-		},
-	}, nil
-}
-
-// send authentication in a header rather than in the stream.
-func createAuthenticationHeader(instanceID string, identity *cryptoutil.PrivateKey) (string, error) {
-	signedMessage, err := createSignedMessage([]byte(instanceID), identity)
-	if err != nil {
-		return "", fmt.Errorf("signing instance ID for authentication header: %w", err)
-	}
-
-	handshakePacket := &v1.SyncStreamItem_SyncActionHandshake{
-		ProtocolVersion: SyncProtocolVersion,
-		InstanceId:      signedMessage,
-		PublicKey:       identity.PublicKeyProto(),
-	}
-
-	encodedHandshake, err := proto.Marshal(handshakePacket)
-	if err != nil {
-		return "", fmt.Errorf("marshalling handshake packet: %w", err)
-	}
-
-	if len(encodedHandshake) > 512 {
-		return "", fmt.Errorf("authorization header is too large, max size is 512 bytes, got %d bytes", len(encodedHandshake))
-	}
-
-	base64Handshake := base64.StdEncoding.EncodeToString(encodedHandshake)
-	return fmt.Sprintf("Backrest-Sync-Auth %s", base64Handshake), nil
-}
-
-// verifyHandshakePacket verifies that
-//   - the signature on the instance ID is valid against the public key provided in the handshake
-//   - that the public key's ID is as attested in the handshake packet e.g. matches handshake.PublicKey.Keyid
-//
-// To authenticate, the caller must then check that the public key is trusted by checking the key ID against a local list.
-func verifyHandshakePacket(item *v1.SyncStreamItem) (*cryptoutil.PublicKey, error) {
-	handshake := item.GetHandshake()
-	if handshake == nil {
-		return nil, fmt.Errorf("empty or nil handshake, handshake packet must be sent first")
-	}
-
-	if handshake.ProtocolVersion != SyncProtocolVersion {
-		return nil, fmt.Errorf("protocol version mismatch: expected %d, got %d", SyncProtocolVersion, handshake.ProtocolVersion)
-	}
-
-	if len(handshake.InstanceId.GetPayload()) == 0 || len(handshake.InstanceId.GetSignature()) == 0 {
-		return nil, errors.New("instance ID payload and signature must not be empty")
-	}
-
-	if len(handshake.PublicKey.Keyid) == 0 {
-		return nil, errors.New("public key ID must not be empty")
-	}
-
-	peerKey, err := cryptoutil.NewPublicKey(handshake.PublicKey)
-	if err != nil {
-		return nil, fmt.Errorf("loading peer public key: %w", err)
-	}
-
-	if err := verifySignedMessage(handshake.InstanceId, peerKey); err != nil {
-		return nil, fmt.Errorf("verifying instance ID signature: %w", err)
-	}
-
-	return peerKey, nil
-}
-
-func verifyAuthenticationHeader(header string) (*cryptoutil.PublicKey, string, error) {
-	if len(header) == 0 {
-		return nil, "", errors.New("authentication header must not be empty")
-	}
-
-	// The header is expected to be in the format "Backrest-Sync-Auth <base64-encoded-handshake-packet>"
-	if !strings.HasPrefix(header, "Backrest-Sync-Auth ") {
-		return nil, "", fmt.Errorf("invalid authentication header format, expected 'Backrest-Sync-Auth <base64-encoded-handshake-packet>', got %s", header)
-	}
-
-	// Extract the base64-encoded handshake packet
-	header = header[len("Backrest-Sync-Auth "):]
-	decoded, err := base64.StdEncoding.DecodeString(header)
-	if err != nil {
-		return nil, "", fmt.Errorf("decoding authentication header: %w", err)
-	}
-
-	// Unmarshal the decoded header into a handshake packet
-	var handshakePacket v1.SyncStreamItem_SyncActionHandshake
-	if err := proto.Unmarshal(decoded, &handshakePacket); err != nil {
-		return nil, "", fmt.Errorf("unmarshalling handshake packet: %w", err)
-	}
-
-	// Verify the handshake packet
-	peerKey, err := verifyHandshakePacket(&v1.SyncStreamItem{
-		Action: &v1.SyncStreamItem_Handshake{
-			Handshake: &handshakePacket,
-		},
-	})
-	if err != nil {
-		return nil, "", fmt.Errorf("verifying handshake packet: %w", err)
-	}
-
-	return peerKey, string(handshakePacket.GetInstanceId().GetPayload()), nil
-}

internal/api/syncapi/cryptoutil.go

@@ -1,68 +0,0 @@
-package syncapi
-
-import (
-	"encoding/binary"
-	"errors"
-	"fmt"
-	"time"
-
-	v1 "github.com/garethgeorge/backrest/gen/go/v1"
-	"github.com/garethgeorge/backrest/internal/cryptoutil"
-)
-
-func createSignedMessage(payload []byte, identity *cryptoutil.PrivateKey) (*v1.SignedMessage, error) {
-	if len(payload) == 0 {
-		return nil, errors.New("payload must not be empty")
-	}
-
-	timestampMillis := time.Now().UnixMilli()
-
-	payloadWithTimestamp := make([]byte, 0, len(payload)+8)
-	binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(timestampMillis))
-	payloadWithTimestamp = append(payloadWithTimestamp, payload...)
-
-	signature, err := identity.Sign(payloadWithTimestamp)
-	if err != nil {
-		return nil, fmt.Errorf("signing payload: %w", err)
-	}
-
-	return &v1.SignedMessage{
-		Payload:         payload,
-		Signature:       signature,
-		Keyid:           identity.KeyID(),
-		TimestampMillis: timestampMillis,
-	}, nil
-}
-
-func verifySignedMessage(msg *v1.SignedMessage, publicKey *cryptoutil.PublicKey) error {
-	if msg == nil {
-		return errors.New("signed message must not be nil")
-	}
-	if len(msg.GetPayload()) == 0 {
-		return errors.New("signed message payload must not be empty")
-	}
-	if len(msg.GetSignature()) == 0 {
-		return errors.New("signed message signature must not be empty")
-	}
-	if len(msg.GetKeyid()) == 0 {
-		return errors.New("signed message key ID must not be empty")
-	}
-
-	if publicKey.KeyID() != msg.GetKeyid() {
-		return fmt.Errorf("public key ID mismatch: expected %s, got %s", publicKey.KeyID(), msg.GetKeyid())
-	}
-
-	payloadWithTimestamp := make([]byte, 0, len(msg.GetPayload())+8)
-	binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(msg.GetTimestampMillis()))
-	payloadWithTimestamp = append(payloadWithTimestamp, msg.GetPayload()...)
-
-	if err := publicKey.Verify(payloadWithTimestamp, msg.GetSignature()); err != nil {
-		return fmt.Errorf("verifying signed message: %w", err)
-	}
-
-	if time.Since(time.UnixMilli(msg.GetTimestampMillis())) > maxSignatureAge {
-		return fmt.Errorf("signature is too old, max age is %s. Is the clock out of sync?", maxSignatureAge)
-	}
-
-	return nil
-}

internal/api/syncapi/errors.go

@@ -42,6 +42,20 @@ func NewSyncErrorPending(message error) *SyncError {
 	}
 }
 
+func NewSyncErrorConnected(message error) *SyncError {
+	return &SyncError{
+		State:   v1.SyncConnectionState_CONNECTION_STATE_CONNECTED,
+		Message: message,
+	}
+}
+
+func NewSyncErrorRetryWait(message error) *SyncError {
+	return &SyncError{
+		State:   v1.SyncConnectionState_CONNECTION_STATE_RETRY_WAIT,
+		Message: message,
+	}
+}
+
 func NewSyncErrorAuth(message error) *SyncError {
 	return &SyncError{
 		State:   v1.SyncConnectionState_CONNECTION_STATE_ERROR_AUTH,

internal/api/syncapi/peerstate.go

@@ -25,8 +25,8 @@ type PeerState struct {
 	ConnectionStateMessage string
 
 	// Plans and repos available on this peer
-	KnownRepos map[string]*v1.SyncRepoMetadata
-	KnownPlans map[string]*v1.SyncPlanMetadata
+	KnownRepos map[string]struct{}
+	KnownPlans map[string]struct{}
 
 	// Partial configuration available for this peer
 	Config *v1.RemoteConfig
@@ -39,8 +39,8 @@ func newPeerState(instanceID, keyID string) *PeerState {
 		LastHeartbeat:          time.Now(),
 		ConnectionState:        v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED,
 		ConnectionStateMessage: "disconnected",
-		KnownRepos:             make(map[string]*v1.SyncRepoMetadata),
-		KnownPlans:             make(map[string]*v1.SyncPlanMetadata),
+		KnownRepos:             make(map[string]struct{}),
+		KnownPlans:             make(map[string]struct{}),
 		Config:                 nil, // Will be set when the config is received
 	}
 }
@@ -68,8 +68,8 @@ func peerStateToProto(state *PeerState) *v1.PeerState {
 		LastHeartbeatMillis: state.LastHeartbeat.UnixMilli(),
 		State:               state.ConnectionState,
 		StatusMessage:       state.ConnectionStateMessage,
-		KnownRepos:          slices.Collect(maps.Values(state.KnownRepos)),
-		KnownPlans:          slices.Collect(maps.Values(state.KnownPlans)),
+		KnownRepos:          slices.Collect(maps.Keys(state.KnownRepos)),
+		KnownPlans:          slices.Collect(maps.Keys(state.KnownPlans)),
 		RemoteConfig:        state.Config,
 	}
 }
@@ -78,13 +78,13 @@ func peerStateFromProto(state *v1.PeerState) *PeerState {
 	if state.PeerInstanceId == "" || state.PeerKeyid == "" {
 		return nil
 	}
-	knownRepos := make(map[string]*v1.SyncRepoMetadata, len(state.KnownRepos))
+	knownRepos := make(map[string]struct{}, len(state.KnownRepos))
 	for _, repo := range state.KnownRepos {
-		knownRepos[repo.Id] = repo
+		knownRepos[repo] = struct{}{}
 	}
-	knownPlans := make(map[string]*v1.SyncPlanMetadata, len(state.KnownPlans))
+	knownPlans := make(map[string]struct{}, len(state.KnownPlans))
 	for _, plan := range state.KnownPlans {
-		knownPlans[plan.Id] = plan
+		knownPlans[plan] = struct{}{}
 	}
 
 	return &PeerState{
@@ -103,7 +103,6 @@ type PeerStateManager interface {
 	GetPeerState(keyID string) *PeerState
 	GetAll() []*PeerState
 	SetPeerState(keyID string, state *PeerState)
-	UpdatePeerState(keyID string, instanceID string, updateFn func(state *PeerState))
 	OnStateChanged() eventemitter.Receiver[*PeerState]
 	Close() error
 }
@@ -159,23 +158,6 @@ func (m *InMemoryPeerStateManager) SetPeerState(keyID string, state *PeerState)
 	m.onStateChanged.Emit(copy)
 }
 
-func (m *InMemoryPeerStateManager) UpdatePeerState(keyID string, instanceID string, updateFn func(state *PeerState)) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	var state *PeerState
-	if existingState, exists := m.peerStates[keyID]; exists {
-		state = existingState.Clone()
-	} else {
-		state = newPeerState(instanceID, keyID)
-	}
-
-	updateFn(state)
-
-	m.peerStates[keyID] = state
-	m.onStateChanged.Emit(state.Clone())
-}
-
 func (m *InMemoryPeerStateManager) Close() error {
 	return nil
 }
@@ -264,41 +246,6 @@ func (m *SqlitePeerStateManager) SetPeerState(keyID string, state *PeerState) {
 	m.onStateChanged.Emit(state.Clone())
 }
 
-func (m *SqlitePeerStateManager) UpdatePeerState(keyID string, instanceID string, updateFn func(state *PeerState)) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	var state *PeerState
-	if stateBytes, err := m.kvstore.Get(keyID); err == nil {
-		var stateProto v1.PeerState
-		if err := proto.Unmarshal(stateBytes, &stateProto); err != nil {
-			zap.S().Warnf("error unmarshalling peer state for key %s: %v", keyID, err)
-		} else {
-			state = peerStateFromProto(&stateProto)
-		}
-	} else {
-		zap.S().Warnf("error getting peer state for key %s: %v", keyID, err)
-	}
-	if state == nil {
-		state = newPeerState(instanceID, keyID)
-	}
-
-	updateFn(state)
-
-	stateProto := peerStateToProto(state)
-	newStateBytes, err := proto.Marshal(stateProto)
-	if err != nil {
-		zap.S().Warnf("error marshalling peer state for key %s: %v", keyID, err)
-		return
-	}
-
-	if err := m.kvstore.Set(keyID, newStateBytes); err != nil {
-		zap.S().Warnf("error setting peer state for key %s: %v", keyID, err)
-		return
-	}
-	m.onStateChanged.Emit(state.Clone())
-}
-
 func (m *SqlitePeerStateManager) Close() error {
 	return nil
 }

internal/api/syncapi/peerstate_test.go

@@ -1,26 +1,15 @@
 package syncapi
 
 import (
-	"database/sql"
 	"testing"
 	"time"
 
-<<<<<<< HEAD
+	"github.com/garethgeorge/backrest/internal/kvstore"
 	"github.com/google/go-cmp/cmp"
-	_ "github.com/ncruces/go-sqlite3/driver"
-	"github.com/ncruces/go-sqlite3/vfs/memdb"
-=======
-	v1 "github.com/garethgeorge/backrest/gen/go/v1"
-	"github.com/garethgeorge/backrest/internal/cryptoutil"
-	"github.com/google/go-cmp/cmp"
-	"google.golang.org/protobuf/testing/protocmp"
-	"zombiezen.com/go/sqlite"
-	"zombiezen.com/go/sqlite/sqlitex"
->>>>>>> 9041d3c (improve sync api security by using 'Authorization' headers for initial key exchange)
 )
 
 func PeerStateManagersForTest(t testing.TB) map[string]PeerStateManager {
-	dbpool := newDbForTest(t)
+	dbpool := kvstore.NewInMemorySqliteDbForKvStore(t)
 	t.Cleanup(func() {
 		dbpool.Close()
 	})
@@ -44,18 +33,12 @@ func TestPeerStateManager_GetSet(t *testing.T) {
 				InstanceID:    "testInstance",
 				KeyID:         keyID,
 				LastHeartbeat: time.Now().Round(time.Millisecond),
-				KnownRepos: map[string]*v1.SyncRepoMetadata{
-					"repo1": {Id: "repo1"},
-					"repo2": {Id: "repo2"},
-				},
-				KnownPlans: map[string]*v1.SyncPlanMetadata{
-					"plan1": {Id: "plan1"},
-					"plan2": {Id: "plan2"},
-				},
+				KnownRepos:    map[string]struct{}{"repo1": {}, "repo2": {}},
+				KnownPlans:    map[string]struct{}{"plan1": {}, "plan2": {}},
 			}
 			psm.SetPeerState(keyID, state)
 			gotState := psm.GetPeerState(keyID)
-			if diff := cmp.Diff(state, gotState, protocmp.Transform()); diff != "" {
+			if diff := cmp.Diff(state, gotState, cmp.AllowUnexported(PeerState{})); diff != "" {
 				t.Errorf("unexpected diff: %v", diff)
 			}
 		})
@@ -95,7 +78,7 @@ func TestPeerStateManager_OnStateChanged(t *testing.T) {
 
 			select {
 			case changedState := <-ch:
-				if diff := cmp.Diff(state, changedState, protocmp.Transform()); diff != "" {
+				if diff := cmp.Diff(state, changedState, cmp.AllowUnexported(PeerState{})); diff != "" {
 					t.Errorf("unexpected diff: %v", diff)
 				}
 			case <-time.After(1 * time.Second):
@@ -104,12 +87,3 @@ func TestPeerStateManager_OnStateChanged(t *testing.T) {
 		})
 	}
 }
-
-func newDbForTest(t testing.TB) *sql.DB {
-	t.Helper()
-	dbpool, err := sql.Open("sqlite3", memdb.TestDB(t))
-	if err != nil {
-		t.Fatalf("error creating sqlite pool: %s", err)
-	}
-	return dbpool
-}

internal/api/syncapi/syncapi_test.go

@@ -539,9 +539,6 @@ func tryExpectExactOperations(t *testing.T, ctx context.Context, peer *peerUnder
 		for _, op := range ops {
 			op.Modno = 0
 		}
-		for _, op := range wantOps {
-			op.Modno = 0
-		}
 		if diff := cmp.Diff(ops, wantOps, protocmp.Transform()); diff != "" {
 			return fmt.Errorf("unexpected diff: %v", diff)
 		}
@@ -626,13 +623,13 @@ func waitForConnectionState(t *testing.T, ctx context.Context, peer *peerUnderTe
 	onStateChanged := peer.manager.peerStateManager.OnStateChanged().Subscribe()
 	defer peer.manager.peerStateManager.OnStateChanged().Unsubscribe(onStateChanged)
 
-	// First check if the peer is already in the desired state.
+	// First check if the peer is already connected.
 	state := peer.manager.peerStateManager.GetPeerState(hostPeer.Keyid)
-	if state != nil && state.ConnectionState == wantState {
-		return // Already in desired state, nothing to do
+	if state != nil && state.ConnectionState == v1.SyncConnectionState_CONNECTION_STATE_CONNECTED {
+		return // Already connected, nothing to do
 	}
 
-	// If not in desired state, wait for a connection event
+	// If not connected, wait for a connection event
 	var lastState *PeerState
 	stop := false
 	for !stop {
@@ -644,7 +641,7 @@ func waitForConnectionState(t *testing.T, ctx context.Context, peer *peerUnderTe
 			}
 			if state.KeyID == hostPeer.Keyid && state.InstanceID == hostPeer.InstanceId {
 				lastState = state
-				if state.ConnectionState == wantState {
+				if state.ConnectionState == v1.SyncConnectionState_CONNECTION_STATE_CONNECTED {
 					stop = true
 					continue
 				}
@@ -656,8 +653,8 @@ func waitForConnectionState(t *testing.T, ctx context.Context, peer *peerUnderTe
 	}
 	if lastState == nil {
 		t.Fatalf("timeout waiting for connection to host peer %s", hostPeer.InstanceId)
-	} else if lastState.ConnectionState != wantState {
-		t.Fatalf("expected connection state to be %v, got %v (reason: %q)", wantState, lastState.ConnectionState, lastState.ConnectionStateMessage)
+	} else if lastState.ConnectionState != v1.SyncConnectionState_CONNECTION_STATE_CONNECTED {
+		t.Fatalf("expected connection state to be CONNECTED, got %v (reason: %q)", lastState.ConnectionState, lastState.ConnectionStateMessage)
 	}
 }
 
@@ -668,8 +665,8 @@ func tryConnect(t *testing.T, ctx context.Context, peer *peerUnderTest, hostPeer
 func runSyncAPIWithCtx(ctx context.Context, peer *peerUnderTest, bindAddr string) {
 	mux := http.NewServeMux()
 	syncHandler := NewBackrestSyncHandler(peer.manager)
-	endPoint, syncServiceHandler := v1connect.NewBackrestSyncServiceHandler(syncHandler)
-	mux.Handle(endPoint, AuthenticationMiddleware(peer.configMgr, syncServiceHandler))
+	mux.Handle(v1connect.NewBackrestSyncServiceHandler(syncHandler))
+
 	server := &http.Server{
 		Addr:    bindAddr,
 		Handler: h2c.NewHandler(mux, &http2.Server{}), // h2c is HTTP/2 without TLS for grpc-connect support.
@@ -768,7 +765,7 @@ func newPeerUnderTest(t *testing.T, initialConfig *v1.Config) *peerUnderTest {
 		t.Fatalf("failed to create peer state manager: %v", err)
 	}
 
-	manager := NewSyncManager(configMgr, oplog, logStore, orchestrator, peerStateManager)
+	manager := NewSyncManager(configMgr, oplog, orchestrator, peerStateManager)
 	manager.syncClientRetryDelay = 250 * time.Millisecond
 
 	return &peerUnderTest{

internal/api/syncapi/syncclient.go

@@ -74,9 +74,7 @@ func NewSyncClient(
 		oplog:              oplog,
 		l:                  zap.L().Named(fmt.Sprintf("syncclient for %q", peer.GetInstanceId())),
 	}
-	c.mgr.peerStateManager.UpdatePeerState(peer.Keyid, peer.InstanceId, func(peerState *PeerState) {
-		// this will create a new peer state if one doesn't exist
-	})
+	c.mgr.peerStateManager.SetPeerState(peer.Keyid, newPeerState(peer.InstanceId, peer.Keyid))
 	return c, nil
 }
 
@@ -101,92 +99,47 @@ func (c *SyncClient) RunSync(ctx context.Context) {
 		c.l.Sugar().Infof("connecting to peer %q (%s) at %s", c.peer.InstanceId, c.peer.Keyid, c.peer.GetInstanceUrl())
 
 		var wg sync.WaitGroup
-
-		// Attach the 'Authorization' header to the sync stream as we create it.
-		syncStream := c.client.Sync(ctx)
-		if authHeader, err := createAuthenticationHeader(c.syncConfigSnapshot.config.Instance, c.syncConfigSnapshot.identityKey); err != nil {
-			c.l.Sugar().Errorf("failed to create authentication header: %v", err)
-		} else {
-			syncStream.RequestHeader().Set("Authorization", authHeader)
-		}
-
-		// Start a thread to handle the stream IO e.g. send and receive messages and pass them into the cmdStream stream abstraction.
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			if err := cmdStream.ConnectStream(ctx, syncStream); err != nil {
-				c.l.Sugar().Infof("lost stream connection to peer %q (%s): %v", c.peer.InstanceId, c.peer.Keyid, err)
-				var syncErr *SyncError
-				c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(state *PeerState) {
-					state.LastHeartbeat = time.Now()
-					if errors.As(err, &syncErr) {
-						state.ConnectionState = syncErr.State
-						state.ConnectionStateMessage = syncErr.Message.Error()
-					} else {
-						state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_ERROR_INTERNAL
-						state.ConnectionStateMessage = err.Error()
-					}
-				})
-			} else {
-				c.l.Sugar().Infof("stream connection to peer %q (%s) closed", c.peer.InstanceId, c.peer.Keyid)
-				c.reconnectAttempts = 0
-				c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(state *PeerState) {
-					state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
-					state.ConnectionStateMessage = "disconnected"
-					state.LastHeartbeat = time.Now()
-				})
-			}
-		}()
-
-		// Start a thread to handle the API loop, this is fundamentally a switch statement operating on incomming commands.
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-
-			// Get the headers from the stream
-			headers := syncStream.ResponseHeader()
-			authHeader := headers.Get("Authorization")
-
-			peerKey, instanceID, err := verifyAuthenticationHeader(authHeader)
-			if err != nil {
-				cmdStream.SendErrorAndTerminate(NewSyncErrorAuth(fmt.Errorf("verifying authentication header: %w", err)))
-				return
-			}
-			if peerKey.KeyID() != c.peer.Keyid {
-				cmdStream.SendErrorAndTerminate(NewSyncErrorAuth(fmt.Errorf("remote peer key ID mismatch: expected %q, got %q", c.peer.Keyid, peerKey.KeyID())))
-				return
-			}
-			if instanceID != c.peer.InstanceId {
-				cmdStream.SendErrorAndTerminate(NewSyncErrorAuth(fmt.Errorf("remote instance ID mismatch: expected %q, got %q", c.peer.InstanceId, instanceID)))
-				return
-			}
-			ctx := ContextWithPeer(context.Background(), c.peer, peerKey)
-
-			// Run sync once the first packet has been received allowing us to read the request headers.
-			// This means it is essential that the server always sends some initial packet, today this is just a heartbeat to start off the connection.
-			err = runSync(
+			err := runSync(
 				ctx,
 				c.localInstanceID,
 				c.syncConfigSnapshot.identityKey,
 				cmdStream,
 				syncSessionHandler,
+				c.syncConfigSnapshot.config.GetMultihost().GetKnownHosts(),
 			)
 			cmdStream.SendErrorAndTerminate(err)
 		}()
 
-		// Send a heartbeat packet to trigger establishing the connection.
-		cmdStream.Send(&v1.SyncStreamItem{
-			Action: &v1.SyncStreamItem_Heartbeat{},
-		})
+		if err := cmdStream.ConnectStream(ctx, c.client.Sync(ctx)); err != nil {
+			c.l.Sugar().Infof("lost stream connection to peer %q (%s): %v", c.peer.InstanceId, c.peer.Keyid, err)
+			var syncErr *SyncError
+			state := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone()
+			if state == nil {
+				state = newPeerState(c.peer.InstanceId, c.peer.Keyid)
+			}
+			state.LastHeartbeat = time.Now()
+			if errors.As(err, &syncErr) {
+				state.ConnectionState = syncErr.State
+				state.ConnectionStateMessage = syncErr.Message.Error()
+			} else {
+				state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_ERROR_INTERNAL
+				state.ConnectionStateMessage = err.Error()
+			}
+			c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, state)
+		} else {
+			c.reconnectAttempts = 0
+		}
 
-		// Wait for the thread running the API loop and the thread running the stream connection to finish.
 		wg.Wait()
 
-		reconnectDelayWithBackoff := c.reconnectDelay
+		delay := c.reconnectDelay - time.Since(lastConnect)
 		if c.reconnectAttempts > 0 {
-			reconnectDelayWithBackoff *= time.Duration(1 << min(c.reconnectAttempts, 5)) // 2^reconnectAttempts, max 32
+			backoff := time.Duration(1<<min(c.reconnectAttempts, 5)) * c.reconnectDelay // 2^reconnectAttempts, max 32
+			delay += backoff
 		}
-		delay := reconnectDelayWithBackoff - time.Since(lastConnect)
 		c.l.Sugar().Infof("disconnected, will retry after %v (attempt %d)", delay, c.reconnectAttempts)
 		c.reconnectAttempts++
 		select {
@@ -253,11 +206,14 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
 	}
 
 	c.l.Sugar().Infof("sync connection established with peer %q (%s)", peer.InstanceId, peer.Keyid)
-	c.mgr.peerStateManager.UpdatePeerState(peer.Keyid, c.peer.InstanceId, func(peerState *PeerState) {
-		peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_CONNECTED
-		peerState.ConnectionStateMessage = "connected"
-		peerState.LastHeartbeat = time.Now()
-	})
+	peerState := c.mgr.peerStateManager.GetPeerState(peer.Keyid).Clone()
+	if peerState == nil {
+		peerState = newPeerState(c.peer.InstanceId, peer.Keyid)
+	}
+	peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_CONNECTED
+	peerState.ConnectionStateMessage = "connected"
+	peerState.LastHeartbeat = time.Now()
+	c.mgr.peerStateManager.SetPeerState(peer.Keyid, peerState)
 
 	// Send a heartbeat every 2 minutes to keep the connection alive.
 	go sendHeartbeats(ctx, stream, env.MultihostHeartbeatInterval())
@@ -274,22 +230,13 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
 		for _, repo := range localConfig.Repos {
 			if c.permissions.CheckPermissionForRepo(repo.Guid, v1.Multihost_Permission_PERMISSION_READ_CONFIG) {
 				remoteConfig.Repos = append(remoteConfig.Repos, repo)
-			}
-			if c.permissions.CheckPermissionForRepo(repo.Guid, v1.Multihost_Permission_PERMISSION_READ_OPERATIONS, v1.Multihost_Permission_PERMISSION_READ_CONFIG) {
-				resourceList.Repos = append(resourceList.Repos, &v1.SyncRepoMetadata{
-					Id:   repo.Id,
-					Guid: repo.Guid,
-				})
+				resourceList.RepoIds = append(resourceList.RepoIds, repo.Id)
 			}
 		}
 		for _, plan := range localConfig.Plans {
 			if c.permissions.CheckPermissionForPlan(plan.Id, v1.Multihost_Permission_PERMISSION_READ_CONFIG) {
 				remoteConfig.Plans = append(remoteConfig.Plans, plan)
-			}
-			if c.permissions.CheckPermissionForPlan(plan.Id, v1.Multihost_Permission_PERMISSION_READ_OPERATIONS, v1.Multihost_Permission_PERMISSION_READ_CONFIG) {
-				resourceList.Plans = append(resourceList.Plans, &v1.SyncPlanMetadata{
-					Id: plan.Id,
-				})
+				resourceList.PlanIds = append(resourceList.PlanIds, plan.Id)
 			}
 		}
 
@@ -362,7 +309,7 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
 	// This is a slow operation and we don't want to block the main loop waiting for it to complete and potentially forcing incomming messages to buffer or drop.
 	go func() {
 		startSync := func(diffSel *v1.OpSelector) error {
-			c.l.Sugar().Debugf("starting sync with diffselector: %v", diffSel)
+			c.l.Sugar().Infof("starting sync with diffselector: %v", diffSel)
 
 			diffQuery, err := protoutil.OpSelectorToQuery(diffSel)
 			if err != nil {
@@ -436,18 +383,13 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
 	return nil
 }
 
-func (c *syncSessionHandlerClient) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error {
-	c.l.Sugar().Infof("syncclient connection closed for client %q", c.peer.GetInstanceId())
-	return nil
-}
-
 func (c *syncSessionHandlerClient) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error {
-	c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(peerState *PeerState) {
-		if peerState == nil {
-			return // this should not happen
-		}
-		peerState.LastHeartbeat = time.Now()
-	})
+	peerState := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone()
+	if peerState == nil {
+		return NewSyncErrorInternal(fmt.Errorf("peer state not found for peer %q", c.peer.InstanceId))
+	}
+	peerState.LastHeartbeat = time.Now()
+	c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, peerState)
 	return nil
 }
 
@@ -530,16 +472,16 @@ func (c *syncSessionHandlerClient) HandleSendOperations(ctx context.Context, str
 
 func (c *syncSessionHandlerClient) HandleSendConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendConfig) error {
 	c.l.Sugar().Debugf("received remote config update")
+	peerState := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone()
+	if peerState == nil {
+		return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", c.peer.Keyid))
+	}
 	newRemoteConfig := item.Config
 	if newRemoteConfig == nil {
 		return NewSyncErrorProtocol(fmt.Errorf("received nil remote config"))
 	}
-	c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(peerState *PeerState) {
-		if peerState == nil {
-			return // this should not happen
-		}
-		peerState.Config = newRemoteConfig
-	})
+	peerState.Config = newRemoteConfig
+	c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, peerState)
 	return nil
 }
 
@@ -636,59 +578,3 @@ func (c *syncSessionHandlerClient) HandleListResources(ctx context.Context, stre
 	c.l.Sugar().Debugf("received ListResources request from peer %q", c.peer.InstanceId)
 	return nil
 }
-
-func (c *syncSessionHandlerClient) HandleGetLog(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionGetLog) error {
-	c.l.Sugar().Debugf("received GetLog request from peer %q for log ID %q", c.peer.InstanceId, item.GetLogId())
-
-	reader, err := c.mgr.logStore.Open(item.GetLogId())
-	if err != nil {
-		c.l.Sugar().Warnf("failed to open log %q: %v", item.GetLogId(), err)
-		return nil
-	}
-
-	// Read everything from the io.ReadCloser respecting context cancellation
-	buf := make([]byte, 128*1024) // 128 KiB buffer, read in large chunks to reduce the number of sends.
-	for {
-		select {
-		case <-ctx.Done():
-			c.l.Sugar().Debugf("context cancelled while reading log %q", item.GetLogId())
-			return nil
-		default:
-
-			// TODO: make this implementation... correct.
-			n, err := reader.Read(buf)
-			if n > 0 {
-				stream.Send(&v1.SyncStreamItem{
-					Action: &v1.SyncStreamItem_SendLogData{
-						SendLogData: &v1.SyncStreamItem_SyncActionSendLogData{
-							LogId: item.GetLogId(),
-							Chunk: buf[:n],
-						},
-					},
-				})
-			}
-			if err != nil {
-				if errors.Is(err, net.ErrClosed) || errors.Is(err, context.Canceled) {
-					c.l.Sugar().Debugf("finished reading log %q, context cancelled or stream closed", item.GetLogId())
-					return nil
-				}
-				c.l.Sugar().Warnf("error reading log %q: %v", item.GetLogId(), err)
-				return fmt.Errorf("reading log %q: %w", item.GetLogId(), err)
-			}
-			if n == 0 {
-				c.l.Sugar().Debugf("finished reading log %q, no more data", item.GetLogId())
-				// No more data to read, we can close the stream.
-				stream.Send(&v1.SyncStreamItem{
-					Action: &v1.SyncStreamItem_SendLogData{
-						SendLogData: &v1.SyncStreamItem_SyncActionSendLogData{
-							LogId: item.GetLogId(),
-							Chunk: nil, // Send an empty chunk to signal the end of the log.
-						},
-					},
-				})
-			}
-		}
-	}
-
-	return nil
-}

internal/api/syncapi/synccommon.go

@@ -2,12 +2,14 @@ package syncapi
 
 import (
 	"context"
+	"encoding/binary"
+	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	v1 "github.com/garethgeorge/backrest/gen/go/v1"
 	"github.com/garethgeorge/backrest/internal/cryptoutil"
-	"go.uber.org/zap"
 )
 
 var maxSignatureAge = 5 * time.Minute
@@ -18,18 +20,41 @@ func runSync(
 	localKey *cryptoutil.PrivateKey,
 	commandStream *bidiSyncCommandStream,
 	handler syncSessionHandler,
+	knownPeers []*v1.Multihost_Peer, // could be known hosts or authorized clients, doesn't matter. This is used to verify the handshake packet, authorization comes later.
 ) error {
-	peer := PeerFromContext(ctx)
-	peerPublicKey := PeerPublicKeyFromContext(ctx)
-	if peer == nil || peerPublicKey == nil {
-		return NewSyncErrorAuth(fmt.Errorf("peer not found in context, ensure authentication middleware is applied before sync handlers"))
+	// send the initial handshake packet to the peer to establish the connection.
+	go func() {
+		handshakePacket, err := createHandshakePacket(localInstanceID, localKey)
+		if err != nil {
+			commandStream.SendErrorAndTerminate(fmt.Errorf("creating handshake packet: %w", err))
+			return
+		}
+		commandStream.Send(handshakePacket)
+	}()
+
+	// Wait for the handshake packet to be acknowledged by the peer.
+	handshake := commandStream.ReceiveWithinDuration(15 * time.Second)
+	if handshake == nil {
+		return NewSyncErrorAuth(fmt.Errorf("no handshake packet received from peer within timeout"))
+	}
+	if _, err := verifyHandshakePacket(handshake); err != nil {
+		return NewSyncErrorAuth(fmt.Errorf("verifying handshake packet: %w", err))
 	}
 
-	defer func() {
-		if err := handler.OnConnectionClosed(ctx, commandStream); err != nil {
-			zap.L().Error("error handling connection closed", zap.Error(err))
-		}
-	}()
+	// Find the peer definition in the known peers list.
+	var peer *v1.Multihost_Peer
+	peerIdx := slices.IndexFunc(knownPeers, func(p *v1.Multihost_Peer) bool {
+		return p.Keyid == handshake.GetHandshake().GetPublicKey().GetKeyid()
+	})
+	if peerIdx >= 0 {
+		peer = knownPeers[peerIdx]
+	} else {
+		return NewSyncErrorAuth(fmt.Errorf("peer public key ID %s (instance ID %s) not found in known peers", handshake.GetHandshake().GetPublicKey().GetKeyid(), string(handshake.GetHandshake().GetInstanceId().GetPayload())))
+	}
+
+	if err := authorizeHandshakeAsPeer(handshake, peer); err != nil {
+		return NewSyncErrorAuth(fmt.Errorf("authorizing handshake as peer: %w", err))
+	}
 
 	if err := handler.OnConnectionEstablished(ctx, commandStream, peer); err != nil {
 		return err
@@ -65,14 +90,6 @@ func runSync(
 			if err := handler.HandleThrottle(ctx, commandStream, item.GetThrottle()); err != nil {
 				return fmt.Errorf("handling throttle: %w", err)
 			}
-		case *v1.SyncStreamItem_GetLog:
-			if err := handler.HandleGetLog(ctx, commandStream, item.GetGetLog()); err != nil {
-				return fmt.Errorf("handling get log: %w", err)
-			}
-		case *v1.SyncStreamItem_SendLogData:
-			if err := handler.HandleSendLogData(ctx, commandStream, item.GetSendLogData()); err != nil {
-				return fmt.Errorf("handling send log data: %w", err)
-			}
 		default:
 			return NewSyncErrorProtocol(fmt.Errorf("unknown action type %T in sync stream item", item.GetAction()))
 		}
@@ -97,6 +114,131 @@ func tryReceiveWithinDuration(ctx context.Context, receiveChan chan *v1.SyncStre
 	}
 }
 
+func createSignedMessage(payload []byte, identity *cryptoutil.PrivateKey) (*v1.SignedMessage, error) {
+	if len(payload) == 0 {
+		return nil, errors.New("payload must not be empty")
+	}
+
+	timestampMillis := time.Now().UnixMilli()
+
+	payloadWithTimestamp := make([]byte, 0, len(payload)+8)
+	binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(timestampMillis))
+	payloadWithTimestamp = append(payloadWithTimestamp, payload...)
+
+	signature, err := identity.Sign(payloadWithTimestamp)
+	if err != nil {
+		return nil, fmt.Errorf("signing payload: %w", err)
+	}
+
+	return &v1.SignedMessage{
+		Payload:         payload,
+		Signature:       signature,
+		Keyid:           identity.KeyID(),
+		TimestampMillis: timestampMillis,
+	}, nil
+}
+
+func verifySignedMessage(msg *v1.SignedMessage, publicKey *cryptoutil.PublicKey) error {
+	if msg == nil {
+		return errors.New("signed message must not be nil")
+	}
+	if len(msg.GetPayload()) == 0 {
+		return errors.New("signed message payload must not be empty")
+	}
+	if len(msg.GetSignature()) == 0 {
+		return errors.New("signed message signature must not be empty")
+	}
+	if len(msg.GetKeyid()) == 0 {
+		return errors.New("signed message key ID must not be empty")
+	}
+
+	if publicKey.KeyID() != msg.GetKeyid() {
+		return fmt.Errorf("public key ID mismatch: expected %s, got %s", publicKey.KeyID(), msg.GetKeyid())
+	}
+
+	payloadWithTimestamp := make([]byte, 0, len(msg.GetPayload())+8)
+	binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(msg.GetTimestampMillis()))
+	payloadWithTimestamp = append(payloadWithTimestamp, msg.GetPayload()...)
+
+	if err := publicKey.Verify(payloadWithTimestamp, msg.GetSignature()); err != nil {
+		return fmt.Errorf("verifying signed message: %w", err)
+	}
+
+	if time.Since(time.UnixMilli(msg.GetTimestampMillis())) > maxSignatureAge {
+		return fmt.Errorf("signature is too old, max age is %s. Is the clock out of sync?", maxSignatureAge)
+	}
+
+	return nil
+}
+
+func createHandshakePacket(instanceID string, identity *cryptoutil.PrivateKey) (*v1.SyncStreamItem, error) {
+	signedMessage, err := createSignedMessage([]byte(instanceID), identity)
+	if err != nil {
+		return nil, fmt.Errorf("signing instance ID: %w", err)
+	}
+
+	return &v1.SyncStreamItem{
+		Action: &v1.SyncStreamItem_Handshake{
+			Handshake: &v1.SyncStreamItem_SyncActionHandshake{
+				ProtocolVersion: SyncProtocolVersion,
+				InstanceId:      signedMessage,
+				PublicKey:       identity.PublicKeyProto(),
+			},
+		},
+	}, nil
+}
+
+// verifyHandshakePacket verifies that
+//   - the signature on the instance ID is valid against the public key provided in the handshake
+//   - that the public key's ID is as attested in the handshake packet e.g. matches handshake.PublicKey.Keyid
+//
+// To authenticate, the caller must then check that the public key is trusted by checking the key ID against a local list.
+func verifyHandshakePacket(item *v1.SyncStreamItem) (*cryptoutil.PublicKey, error) {
+	handshake := item.GetHandshake()
+	if handshake == nil {
+		return nil, fmt.Errorf("empty or nil handshake, handshake packet must be sent first")
+	}
+
+	if handshake.ProtocolVersion != SyncProtocolVersion {
+		return nil, fmt.Errorf("protocol version mismatch: expected %d, got %d", SyncProtocolVersion, handshake.ProtocolVersion)
+	}
+
+	if len(handshake.InstanceId.GetPayload()) == 0 || len(handshake.InstanceId.GetSignature()) == 0 {
+		return nil, errors.New("instance ID payload and signature must not be empty")
+	}
+
+	if len(handshake.PublicKey.Keyid) == 0 {
+		return nil, errors.New("public key ID must not be empty")
+	}
+
+	peerKey, err := cryptoutil.NewPublicKey(handshake.PublicKey)
+	if err != nil {
+		return nil, fmt.Errorf("loading peer public key: %w", err)
+	}
+
+	if err := verifySignedMessage(handshake.InstanceId, peerKey); err != nil {
+		return nil, fmt.Errorf("verifying instance ID signature: %w", err)
+	}
+
+	return peerKey, nil
+}
+
+// authorizeHandshakeAsPeer checks that the handshake packet has the expected key ID and instance ID.
+// If this succeeds and the handshake is verified, then it is safe to assume the identity we are talking to.
+func authorizeHandshakeAsPeer(item *v1.SyncStreamItem, peer *v1.Multihost_Peer) error {
+	handshake := item.GetHandshake()
+	if handshake == nil {
+		return fmt.Errorf("empty or nil handshake, handshake packet must be sent first")
+	}
+	if string(handshake.GetInstanceId().GetPayload()) != peer.InstanceId {
+		return fmt.Errorf("instance ID mismatch: expected %s, got %s", peer.InstanceId, string(handshake.InstanceId.GetPayload()))
+	}
+	if handshake.GetPublicKey().GetKeyid() != peer.Keyid {
+		return fmt.Errorf("public key ID mismatch: expected %s, got %s", peer.Keyid, handshake.PublicKey.Keyid)
+	}
+	return nil
+}
+
 // sendHeartbeats sends a heartbeat message to the stream at regular intervals.
 // This is useful for keeping the connection alive and ensuring that the peer is still responsive.
 func sendHeartbeats(ctx context.Context, stream *bidiSyncCommandStream, interval time.Duration) {
@@ -120,11 +262,7 @@ func sendHeartbeats(ctx context.Context, stream *bidiSyncCommandStream, interval
 // syncSessionHandler is a stateful handler for the messages within the context of a sync stream session.
 // the handler does not need to be thread safe as it is guaranteed to be called from a single thread.
 type syncSessionHandler interface {
-	// OnConnectionEstablished is called when a new sync connection is established, provides the peer information once identified.
 	OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error
-	// OnConnectionClosed is called when the sync connection is closed, e.g. when the peer disconnects. Guaranteed to be called after OnConnectionEstablished.
-	OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error
-	// Handle* methods are called for each action type in the sync stream.
 	HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error
 	HandleDiffOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionDiffOperations) error
 	HandleSendOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendOperations) error
@@ -132,18 +270,12 @@ type syncSessionHandler interface {
 	HandleSetConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSetConfig) error
 	HandleListResources(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionListResources) error
 	HandleThrottle(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionThrottle) error
-	HandleGetLog(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionGetLog) error
-	HandleSendLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendLogData) error
 }
 
 type unimplementedSyncSessionHandler struct{}
 
 func (h *unimplementedSyncSessionHandler) OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error {
-	return nil // no-op by default.
-}
-
-func (h *unimplementedSyncSessionHandler) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error {
-	panic("must not be unimplemented")
+	return NewSyncErrorProtocol(fmt.Errorf("OnConnectionEstablished not implemented"))
 }
 
 func (h *unimplementedSyncSessionHandler) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error {
@@ -173,11 +305,3 @@ func (h *unimplementedSyncSessionHandler) HandleListResources(ctx context.Contex
 func (h *unimplementedSyncSessionHandler) HandleThrottle(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionThrottle) error {
 	return NewSyncErrorProtocol(fmt.Errorf("HandleThrottle not implemented"))
 }
-
-func (h *unimplementedSyncSessionHandler) HandleGetLog(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionGetLog) error {
-	return NewSyncErrorProtocol(fmt.Errorf("HandleGetLog not implemented"))
-}
-
-func (h *unimplementedSyncSessionHandler) HandleSendLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendLogData) error {
-	return NewSyncErrorProtocol(fmt.Errorf("HandleSendLogData not implemented"))
-}

internal/api/syncapi/synchandler.go

@@ -4,10 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"slices"
 	"sort"
-	"sync"
 	"time"
 
 	"connectrpc.com/connect"
@@ -26,8 +24,6 @@ const SyncProtocolVersion = 1
 type BackrestSyncHandler struct {
 	v1connect.UnimplementedBackrestSyncServiceHandler
 	mgr *SyncManager
-
-	mu sync.Mutex
 }
 
 var _ v1connect.BackrestSyncServiceHandler = &BackrestSyncHandler{}
@@ -46,25 +42,9 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 		return connect.NewError(connect.CodePermissionDenied, errors.New("sync server is not configured"))
 	}
 
-	// Attach the 'Authorization' header to the sync stream as we create it.
-	if authHeader, err := createAuthenticationHeader(snapshot.config.Instance, snapshot.identityKey); err != nil {
-		zap.S().Errorf("failed to create authentication header: %v", err)
-		return connect.NewError(connect.CodeInternal, fmt.Errorf("creating authentication header: %w", err))
-	} else {
-		stream.ResponseHeader().Set("Authorization", authHeader)
-	}
-
-	// Setup to read from the stream and handle commands
-	// Note that runSync will perform the authentication check, which in this case is asserting that a peer is present.
-	// The peer must be provided in the context by the authentication middleware.
 	sessionHandler := newSyncHandlerServer(h.mgr, snapshot)
 	cmdStream := newBidiSyncCommandStream()
 
-	// Send a heartbeat packet to send the initial headers to the client and establish the connection.
-	cmdStream.Send(&v1.SyncStreamItem{
-		Action: &v1.SyncStreamItem_Heartbeat{},
-	})
-
 	go func() {
 		err := runSync(
 			ctx,
@@ -72,29 +52,25 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 			snapshot.identityKey,
 			cmdStream,
 			sessionHandler,
+			snapshot.config.GetMultihost().GetAuthorizedClients(),
 		)
 		cmdStream.SendErrorAndTerminate(err)
 	}()
 
 	if err := cmdStream.ConnectStream(ctx, stream); err != nil {
-		if sessionHandler.peer != nil {
-			zap.S().Errorf("sync handler stream error for client %q: %v", sessionHandler.peer.InstanceId, err)
-			h.mgr.peerStateManager.UpdatePeerState(sessionHandler.peer.Keyid, sessionHandler.peer.InstanceId, func(peerState *PeerState) {
-				peerState.LastHeartbeat = time.Now()
-				peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
-				peerState.ConnectionStateMessage = err.Error()
-				var syncErr *SyncError
-				if errors.As(err, &syncErr) {
-					peerState.ConnectionState = syncErr.State
-					peerState.ConnectionStateMessage = syncErr.Message.Error()
-				}
-			})
-		} else {
-			zap.S().Errorf("sync handler stream error for unestablished session: %v", err)
-		}
-
+		zap.S().Errorf("sync handler stream error: %v", err)
 		var syncErr *SyncError
 		if errors.As(err, &syncErr) {
+			if sessionHandler.peer != nil {
+				peerState := h.mgr.peerStateManager.GetPeerState(sessionHandler.peer.Keyid).Clone()
+				if peerState == nil {
+					peerState = newPeerState(sessionHandler.peer.InstanceId, sessionHandler.peer.Keyid)
+				}
+				peerState.ConnectionState = syncErr.State
+				peerState.ConnectionStateMessage = syncErr.Message.Error()
+				peerState.LastHeartbeat = time.Now()
+				h.mgr.peerStateManager.SetPeerState(sessionHandler.peer.Keyid, peerState)
+			}
 			switch syncErr.State {
 			case v1.SyncConnectionState_CONNECTION_STATE_ERROR_AUTH:
 				return connect.NewError(connect.CodePermissionDenied, syncErr.Message)
@@ -104,19 +80,12 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 				return connect.NewError(connect.CodeInternal, syncErr.Message)
 			}
 		}
-		return connect.NewError(connect.CodeInternal, err)
-	} else {
-		h.mgr.peerStateManager.UpdatePeerState(sessionHandler.peer.Keyid, sessionHandler.peer.InstanceId, func(peerState *PeerState) {
-			peerState.LastHeartbeat = time.Now()
-			peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
-			peerState.ConnectionStateMessage = "disconnected"
-		})
-		zap.S().Infof("sync handler stream closed for client %q", sessionHandler.peer.InstanceId)
 	}
 
 	return nil
 }
 
+// syncSessionHandlerServer is a syncSessionHandler implementation for servers.
 type syncSessionHandlerServer struct {
 	unimplementedSyncSessionHandler
 
@@ -129,9 +98,6 @@ type syncSessionHandlerServer struct {
 	opIDLru   *lru.Cache[int64, int64] // original ID -> local ID
 	flowIDLru *lru.Cache[int64, int64] // original flow ID -> local flow ID
 
-	requestedLogStreams map[string]struct{}
-	activeLogStreams    map[string]io.WriteCloser
-
 	configWatchCh chan struct{} // Channel for configuration updates
 }
 
@@ -144,24 +110,12 @@ func newSyncHandlerServer(mgr *SyncManager, snapshot *syncConfigSnapshot) *syncS
 		snapshot:  *snapshot,
 		opIDLru:   opIDLru,
 		flowIDLru: flowIDLru,
-
-		requestedLogStreams: make(map[string]struct{}),
-		activeLogStreams:    make(map[string]io.WriteCloser),
 	}
 }
 
 var _ syncSessionHandler = (*syncSessionHandlerServer)(nil)
 
 func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error {
-	// Check if the peer is already connected, and then store the connection.
-	h.mgr.mu.Lock()
-	if _, exists := h.mgr.sessionHandlerMap[peer.Keyid]; exists {
-		h.mgr.mu.Unlock()
-		return NewSyncErrorAuth(fmt.Errorf("client %q is already connected", peer.InstanceId))
-	}
-	h.mgr.sessionHandlerMap[peer.Keyid] = h
-	h.mgr.mu.Unlock()
-
 	// Verify that the peer is in our authorized clients list
 	authorizedClientPeerIdx := slices.IndexFunc(h.snapshot.config.Multihost.GetAuthorizedClients(), func(p *v1.Multihost_Peer) bool {
 		return p.InstanceId == peer.InstanceId && p.Keyid == peer.Keyid
@@ -185,11 +139,11 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context,
 	}
 
 	// Configure the state for the connected peer.
-	h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, peer.InstanceId, func(peerState *PeerState) {
-		peerState.ConnectionStateMessage = "connected"
-		peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_CONNECTED
-		peerState.LastHeartbeat = time.Now()
-	})
+	peerState := newPeerState(peer.InstanceId, h.peer.Keyid)
+	peerState.ConnectionStateMessage = "connected"
+	peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_CONNECTED
+	peerState.LastHeartbeat = time.Now()
+	h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState)
 
 	zap.S().Infof("syncserver accepted a connection from client instance ID %q", h.peer.InstanceId)
 
@@ -220,38 +174,16 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context,
 	}()
 
 	// Send initial configuration to client
-	if err := h.sendConfigToClient(stream, h.snapshot.config); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (h *syncSessionHandlerServer) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error {
-	if h.peer != nil {
-		zap.S().Infof("syncserver connection closed for client %q", h.peer.InstanceId)
-		h.mgr.mu.Lock()
-		delete(h.mgr.sessionHandlerMap, h.peer.Keyid)
-		h.mgr.mu.Unlock()
-	}
-
-	// Close any active resources e.g. sinks for active log streams.
-	for logID, logSink := range h.activeLogStreams {
-		if err := logSink.Close(); err != nil {
-			return fmt.Errorf("action SendLogData: closing log stream %q: %w", logID, err)
-		}
-	}
-
-	return nil
+	return h.sendConfigToClient(stream, h.snapshot.config)
 }
 
 func (h *syncSessionHandlerServer) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error {
-	h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) {
-		if peerState == nil {
-			return // this should not happen
-		}
-		peerState.LastHeartbeat = time.Now()
-	})
+	peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone()
+	if peerState == nil {
+		return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", h.peer.Keyid))
+	}
+	peerState.LastHeartbeat = time.Now()
+	h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState)
 	return nil
 }
 
@@ -369,7 +301,7 @@ func (h *syncSessionHandlerServer) HandleDiffOperations(ctx context.Context, str
 func (h *syncSessionHandlerServer) HandleSendOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendOperations) error {
 	switch event := item.GetEvent().Event.(type) {
 	case *v1.OperationEvent_CreatedOperations:
-		zap.L().Debug("syncserver received create operations", zap.Any("operations", event.CreatedOperations.GetOperations()))
+		zap.L().Debug("syncserver received created operations", zap.Any("operations", event.CreatedOperations.GetOperations()))
 		for _, op := range event.CreatedOperations.GetOperations() {
 			if err := h.insertOrUpdate(op); err != nil {
 				return fmt.Errorf("action SendOperations: operation event create %+v: %w", op, err)
@@ -397,79 +329,32 @@ func (h *syncSessionHandlerServer) HandleSendOperations(ctx context.Context, str
 }
 
 func (h *syncSessionHandlerServer) HandleSendConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendConfig) error {
-	h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) {
-		if peerState == nil {
-			return // this should not happen
-		}
-		peerState.Config = item.GetConfig()
-	})
+	peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone()
+	if peerState == nil {
+		return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", h.peer.Keyid))
+	}
+	peerState.Config = item.GetConfig()
+	h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState)
 	return nil
 }
 
 func (h *syncSessionHandlerServer) HandleListResources(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionListResources) error {
 	zap.L().Debug("syncserver received resource list from client", zap.String("client_instance_id", h.peer.InstanceId),
-		zap.Any("repos", item.GetRepos()),
-		zap.Any("plans", item.GetPlans()))
-	h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) {
-		if peerState == nil {
-			return // this should not happen
-		}
-
-		repos := item.GetRepos()
-		plans := item.GetPlans()
-		for _, repo := range repos {
-			peerState.KnownRepos[repo.Id] = repo
-		}
-		for _, plan := range plans {
-			peerState.KnownPlans[plan.Id] = plan
-		}
-	})
-	return nil
-}
-
-func (h *syncSessionHandlerServer) HandleSendLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendLogData) error {
-	h.mgr.mu.Lock()
-	defer h.mgr.mu.Unlock()
-
-	logID := item.GetLogId()
-	if logID == "" {
-		return NewSyncErrorProtocol(errors.New("action SendLogData: log ID is required"))
+		zap.Any("repos", item.GetRepoIds()),
+		zap.Any("plans", item.GetPlanIds()))
+	peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone()
+	if peerState == nil {
+		return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", h.peer.Keyid))
 	}
-	if _, exists := h.requestedLogStreams[logID]; !exists {
-		return NewSyncErrorProtocol(fmt.Errorf("action SendLogData: log ID %q was not requested", logID))
+	repos := item.GetRepoIds()
+	plans := item.GetPlanIds()
+	for _, repoID := range repos {
+		peerState.KnownRepos[repoID] = struct{}{}
 	}
-
-	var logSink io.WriteCloser
-	if s, exists := h.activeLogStreams[logID]; !exists {
-		// Check if there are too many active log streams
-		if len(h.activeLogStreams) >= 16 {
-			return NewSyncErrorProtocol(fmt.Errorf("action SendLogData: too many active log streams, limit is 16"))
-		}
-
-		// If the log stream is not active, we need to create a new one.
-		f, err := h.mgr.logStore.Create(logID, 0, 24*3600) // 24 hour retention, will just be re-requested next time it's wanted.
-		if err != nil {
-			return fmt.Errorf("action SendLogData: creating log stream %q: %w", logID, err)
-		}
-		logSink = f
-		h.activeLogStreams[logID] = logSink
-	} else {
-		logSink = s
+	for _, planID := range plans {
+		peerState.KnownPlans[planID] = struct{}{}
 	}
-
-	if len(item.GetChunk()) == 0 {
-		delete(h.activeLogStreams, logID)
-		delete(h.requestedLogStreams, logID)
-		if err := logSink.Close(); err != nil {
-			return fmt.Errorf("action SendLogData: closing log stream %q: %w", logID, err)
-		}
-		return nil
-	}
-
-	if _, err := logSink.Write(item.GetChunk()); err != nil {
-		return fmt.Errorf("action SendLogData: writing to log stream %q: %w", logID, err)
-	}
-
+	h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState)
 	return nil
 }
 
@@ -545,14 +430,14 @@ func (h *syncSessionHandlerServer) sendConfigToClient(stream *bidiSyncCommandStr
 	for _, repo := range config.Repos {
 		if h.permissions.CheckPermissionForRepo(repo.Id, v1.Multihost_Permission_PERMISSION_READ_CONFIG) {
 			remoteConfig.Repos = append(remoteConfig.Repos, repo)
-			resourceListMsg.Repos = append(resourceListMsg.Repos, &v1.SyncRepoMetadata{Id: repo.Id, Guid: repo.Guid})
+			resourceListMsg.RepoIds = append(resourceListMsg.RepoIds, repo.Id)
 			allowedRepoIDs = append(allowedRepoIDs, repo.Id)
 		}
 	}
 	for _, plan := range config.Plans {
 		if h.permissions.CheckPermissionForPlan(plan.Id, v1.Multihost_Permission_PERMISSION_READ_CONFIG) {
 			remoteConfig.Plans = append(remoteConfig.Plans, plan)
-			resourceListMsg.Plans = append(resourceListMsg.Plans, &v1.SyncPlanMetadata{Id: plan.Id})
+			resourceListMsg.PlanIds = append(resourceListMsg.PlanIds, plan.Id)
 			allowedPlanIDs = append(allowedPlanIDs, plan.Id)
 		}
 	}

internal/api/syncapi/syncmanager.go

@@ -11,7 +11,6 @@ import (
 	v1 "github.com/garethgeorge/backrest/gen/go/v1"
 	"github.com/garethgeorge/backrest/internal/config"
 	"github.com/garethgeorge/backrest/internal/cryptoutil"
-	"github.com/garethgeorge/backrest/internal/logstore"
 	"github.com/garethgeorge/backrest/internal/oplog"
 	"github.com/garethgeorge/backrest/internal/orchestrator"
 	"go.uber.org/zap"
@@ -21,7 +20,6 @@ type SyncManager struct {
 	configMgr    *config.ConfigManager
 	orchestrator *orchestrator.Orchestrator
 	oplog        *oplog.OpLog
-	logStore     *logstore.LogStore
 
 	// mutable properties
 	mu sync.Mutex
@@ -30,27 +28,32 @@ type SyncManager struct {
 
 	syncClientRetryDelay time.Duration // the default retry delay for sync clients, protected by mu
 
-	syncClients       map[string]*SyncClient               // current sync clients, protected by mu
-	sessionHandlerMap map[string]*syncSessionHandlerServer // handlers for sync sessions keyed by peer KeyID, protected by mu
+	syncClients map[string]*SyncClient // current sync clients, protected by mu
 
 	peerStateManager PeerStateManager
 }
 
-func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, logStore *logstore.LogStore, orchestrator *orchestrator.Orchestrator, peerStateManager PeerStateManager) *SyncManager {
+func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, orchestrator *orchestrator.Orchestrator, peerStateManager PeerStateManager) *SyncManager {
 	// Fetch the config, and mark all sync clients and known hosts as disconnected (but preserve other fields).
 	config, err := configMgr.Get()
 	if err == nil {
 		for _, knownHostPeer := range config.GetMultihost().GetKnownHosts() {
-			peerStateManager.UpdatePeerState(knownHostPeer.Keyid, knownHostPeer.InstanceId, func(state *PeerState) {
-				state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
-				state.ConnectionStateMessage = "disconnected"
-			})
+			state := peerStateManager.GetPeerState(knownHostPeer.Keyid).Clone()
+			if state == nil {
+				state = newPeerState(knownHostPeer.InstanceId, knownHostPeer.Keyid)
+			}
+			state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
+			state.ConnectionStateMessage = "disconnected"
+			peerStateManager.SetPeerState(knownHostPeer.Keyid, state)
 		}
 		for _, authorizedClient := range config.GetMultihost().GetAuthorizedClients() {
-			peerStateManager.UpdatePeerState(authorizedClient.Keyid, authorizedClient.InstanceId, func(state *PeerState) {
-				state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
-				state.ConnectionStateMessage = "disconnected"
-			})
+			state := peerStateManager.GetPeerState(authorizedClient.Keyid).Clone()
+			if state == nil {
+				state = newPeerState(authorizedClient.InstanceId, authorizedClient.Keyid)
+			}
+			state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
+			state.ConnectionStateMessage = "disconnected"
+			peerStateManager.SetPeerState(authorizedClient.Keyid, state)
 		}
 	} else {
 		zap.S().Errorf("syncmanager failed to get initial config: %v", err)
@@ -59,11 +62,9 @@ func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, logStor
 		configMgr:    configMgr,
 		orchestrator: orchestrator,
 		oplog:        oplog,
-		logStore:     logStore,
 
 		syncClientRetryDelay: 60 * time.Second,
 		syncClients:          make(map[string]*SyncClient),
-		sessionHandlerMap:    make(map[string]*syncSessionHandlerServer),
 
 		peerStateManager: peerStateManager,
 	}
@@ -151,7 +152,6 @@ func (m *SyncManager) RunSync(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			zap.S().Debugf("syncmanager context canceled for instance %q, stopping sync", m.snapshot.config.GetInstance())
 			return
 		case <-configWatchCh:
 			runSyncWithNewConfig()

internal/kvstore/sqlitedb.go

@@ -42,10 +42,10 @@ func NewSqliteDbForKvStore(db string) (*sql.DB, error) {
 	return dbpool, nil
 }
 
-func NewInMemorySqliteDbForKvStore(t testing.TB) (*sql.DB, error) {
+func NewInMemorySqliteDbForKvStore(t testing.TB) *sql.DB {
 	dbpool, err := sql.Open("sqlite3", memdb.TestDB(t))
 	if err != nil {
-		return nil, fmt.Errorf("open sqlite pool: %v", err)
+		t.Fatalf("failed to open db: %v", err)
 	}
-	return dbpool, nil
+	return dbpool
 }

internal/logstore/logstore.go

@@ -270,23 +270,23 @@ func (ls *LogStore) SelectAll(f func(id string, parentID int64)) error {
 
 // Find logs owned by a specific operation ID.
 func (ls *LogStore) FindLogsWithParent(parentOpID int64) ([]string, error) {
-	conn, err := ls.dbpool.Take(context.Background())
+	rows, err := ls.dbpool.QueryContext(context.Background(), "SELECT id FROM logs WHERE owner_opid = ?", parentOpID)
 	if err != nil {
-		return nil, fmt.Errorf("take connection: %v", err)
-	}
-	defer ls.dbpool.Put(conn)
-
-	var logs []string
-	if err := sqlitex.Execute(conn, "SELECT id FROM logs WHERE owner_opid = ?", &sqlitex.ExecOptions{
-		Args: []any{parentOpID},
-		ResultFunc: func(stmt *sqlite.Stmt) error {
-			logs = append(logs, stmt.ColumnText(0))
-			return nil
-		},
-	}); err != nil {
 		return nil, fmt.Errorf("select logs: %v", err)
 	}
+	defer rows.Close()
 
+	var logs []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, fmt.Errorf("scan log id: %v", err)
+		}
+		logs = append(logs, id)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate logs: %v", err)
+	}
 	return logs, nil
 }