lee/backrest
1
0
Fork 0
mirror of https://github.com/garethgeorge/backrest.git synced 2026-09-27 02:15:34 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix: cont'd work on sync implementation

Browse Source
This commit is contained in:
Gareth George
2025-10-31 18:04:34 -07:00
committed by Gareth
parent 5585c207c9
commit 74eb15df04
4 changed files with 153 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/syncapi/peerstate.go
+55
View File
@@ -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
}
internal/api/syncapi/syncclient.go
+35 -34
View File
@@ -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
}
internal/api/syncapi/synchandler.go
+55 -41
View File
@@ -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
}
internal/api/syncapi/syncmanager.go
+8 -14
View File
@@ -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)