From 74eb15df04228dad623ff626d9e89c1b6b758105 Mon Sep 17 00:00:00 2001 From: Gareth George Date: Mon, 30 Jun 2025 00:01:33 -0700 Subject: [PATCH] fix: cont'd work on sync implementation --- internal/api/syncapi/peerstate.go | 55 +++++++++++++++++ internal/api/syncapi/syncclient.go | 69 +++++++++++---------- internal/api/syncapi/synchandler.go | 96 +++++++++++++++++------------ internal/api/syncapi/syncmanager.go | 22 +++---- 4 files changed, 153 insertions(+), 89 deletions(-) diff --git a/internal/api/syncapi/peerstate.go b/internal/api/syncapi/peerstate.go index a00fd8fe..c5d9744b 100644 --- a/internal/api/syncapi/peerstate.go +++ b/internal/api/syncapi/peerstate.go @@ -103,6 +103,7 @@ 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 } @@ -158,6 +159,23 @@ 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 } @@ -246,6 +264,43 @@ 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 + stateBytes, err := m.kvstore.Get(keyID) + if err != nil { + zap.S().Warnf("error getting peer state for key %s: %v", keyID, err) + state = newPeerState(instanceID, keyID) + } else if stateBytes == nil { + state = newPeerState(instanceID, keyID) + } else { + 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) + state = newPeerState(instanceID, keyID) + } else { + state = peerStateFromProto(&stateProto) + } + } + + 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 } diff --git a/internal/api/syncapi/syncclient.go b/internal/api/syncapi/syncclient.go index 6083d80f..fc6a73fd 100644 --- a/internal/api/syncapi/syncclient.go +++ b/internal/api/syncapi/syncclient.go @@ -74,7 +74,9 @@ func NewSyncClient( oplog: oplog, l: zap.L().Named(fmt.Sprintf("syncclient for %q", peer.GetInstanceId())), } - c.mgr.peerStateManager.SetPeerState(peer.Keyid, newPeerState(peer.InstanceId, peer.Keyid)) + c.mgr.peerStateManager.UpdatePeerState(peer.Keyid, peer.InstanceId, func(peerState *PeerState) { + // this will create a new peer state if one doesn't exist + }) return c, nil } @@ -116,21 +118,23 @@ func (c *SyncClient) RunSync(ctx context.Context) { 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) + 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.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() + }) } wg.Wait() @@ -206,14 +210,11 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context, } c.l.Sugar().Infof("sync connection established with peer %q (%s)", peer.InstanceId, peer.Keyid) - 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) + 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() + }) // Send a heartbeat every 2 minutes to keep the connection alive. go sendHeartbeats(ctx, stream, env.MultihostHeartbeatInterval()) @@ -384,12 +385,12 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context, } func (c *syncSessionHandlerClient) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error { - 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) + 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() + }) return nil } @@ -472,16 +473,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")) } - peerState.Config = newRemoteConfig - c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, peerState) + c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(peerState *PeerState) { + if peerState == nil { + return // this should not happen + } + peerState.Config = newRemoteConfig + }) return nil } diff --git a/internal/api/syncapi/synchandler.go b/internal/api/syncapi/synchandler.go index 279a1f09..c159d734 100644 --- a/internal/api/syncapi/synchandler.go +++ b/internal/api/syncapi/synchandler.go @@ -58,19 +58,24 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre }() if err := cmdStream.ConnectStream(ctx, stream); err != nil { - zap.S().Errorf("sync handler stream error: %v", err) + 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) + } + 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) @@ -80,6 +85,14 @@ 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 @@ -139,11 +152,11 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context, } // Configure the state for the connected peer. - 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) + 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() + }) zap.S().Infof("syncserver accepted a connection from client instance ID %q", h.peer.InstanceId) @@ -178,12 +191,12 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context, } func (h *syncSessionHandlerServer) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error { - 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) + 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() + }) return nil } @@ -329,12 +342,12 @@ func (h *syncSessionHandlerServer) HandleSendOperations(ctx context.Context, str } func (h *syncSessionHandlerServer) HandleSendConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendConfig) error { - 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) + 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() + }) return nil } @@ -342,19 +355,20 @@ func (h *syncSessionHandlerServer) HandleListResources(ctx context.Context, stre zap.L().Debug("syncserver received resource list from client", zap.String("client_instance_id", h.peer.InstanceId), 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)) - } - repos := item.GetRepoIds() - plans := item.GetPlanIds() - for _, repoID := range repos { - peerState.KnownRepos[repoID] = struct{}{} - } - for _, planID := range plans { - peerState.KnownPlans[planID] = struct{}{} - } - h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState) + h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) { + if peerState == nil { + return // this should not happen + } + + repos := item.GetRepoIds() + plans := item.GetPlanIds() + for _, repoID := range repos { + peerState.KnownRepos[repoID] = struct{}{} + } + for _, planID := range plans { + peerState.KnownPlans[planID] = struct{}{} + } + }) return nil } diff --git a/internal/api/syncapi/syncmanager.go b/internal/api/syncapi/syncmanager.go index 550156c0..af8828c1 100644 --- a/internal/api/syncapi/syncmanager.go +++ b/internal/api/syncapi/syncmanager.go @@ -38,22 +38,16 @@ func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, orchest config, err := configMgr.Get() if err == nil { for _, knownHostPeer := range config.GetMultihost().GetKnownHosts() { - 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) + peerStateManager.UpdatePeerState(knownHostPeer.Keyid, knownHostPeer.InstanceId, func(state *PeerState) { + state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED + state.ConnectionStateMessage = "disconnected" + }) } for _, authorizedClient := range config.GetMultihost().GetAuthorizedClients() { - 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) + peerStateManager.UpdatePeerState(authorizedClient.Keyid, authorizedClient.InstanceId, func(state *PeerState) { + state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED + state.ConnectionStateMessage = "disconnected" + }) } } else { zap.S().Errorf("syncmanager failed to get initial config: %v", err)