fix: cont'd work on sync implementation

Author: garethgeorge

internal/api/syncapi/syncclient.go

@@ -131,6 +131,14 @@ func (c *SyncClient) RunSync(ctx context.Context) {
 			c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, state)
 		} else {
 			c.reconnectAttempts = 0
+			state := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone()
+			if state == nil {
+				state = newPeerState(c.peer.InstanceId, c.peer.Keyid)
+			}
+			state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
+			state.ConnectionStateMessage = "disconnected"
+			state.LastHeartbeat = time.Now()
+			c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, state)
 		}
 
 		wg.Wait()

internal/api/syncapi/synchandler.go

@@ -58,19 +58,27 @@ 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)
-		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)
-				}
+		if sessionHandler.peer != nil {
+			zap.S().Errorf("sync handler stream error for client %q: %v", sessionHandler.peer.InstanceId, err)
+			peerState := h.mgr.peerStateManager.GetPeerState(sessionHandler.peer.Keyid).Clone()
+			if peerState == nil {
+				peerState = newPeerState(sessionHandler.peer.InstanceId, sessionHandler.peer.Keyid)
+			}
+			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()
-				peerState.LastHeartbeat = time.Now()
-				h.mgr.peerStateManager.SetPeerState(sessionHandler.peer.Keyid, peerState)
 			}
+			h.mgr.peerStateManager.SetPeerState(sessionHandler.peer.Keyid, peerState)
+		} else {
+			zap.S().Errorf("sync handler stream error for unestablished session: %v", err)
+		}
+
+		var syncErr *SyncError
+		if errors.As(err, &syncErr) {
 			switch syncErr.State {
 			case v1.SyncConnectionState_CONNECTION_STATE_ERROR_AUTH:
 				return connect.NewError(connect.CodePermissionDenied, syncErr.Message)
@@ -80,6 +88,17 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 				return connect.NewError(connect.CodeInternal, syncErr.Message)
 			}
 		}
+		return connect.NewError(connect.CodeInternal, err)
+	} else {
+		peerState := h.mgr.peerStateManager.GetPeerState(sessionHandler.peer.Keyid).Clone()
+		if peerState == nil {
+			peerState = newPeerState(sessionHandler.peer.InstanceId, sessionHandler.peer.Keyid)
+		}
+		peerState.LastHeartbeat = time.Now()
+		peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED
+		peerState.ConnectionStateMessage = "disconnected"
+		h.mgr.peerStateManager.SetPeerState(sessionHandler.peer.Keyid, peerState)
+		zap.S().Infof("sync handler stream closed for client %q", sessionHandler.peer.InstanceId)
 	}
 
 	return nil