lee/backrest
1
0
Fork 0
mirror of https://github.com/garethgeorge/backrest.git synced 2026-08-26 10:56:33 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

misc bug fixes, add lock hooks, dashboard improvements

Browse Source
This commit is contained in:
Gareth George
2026-05-02 22:27:28 -07:00
parent bdbd2b46c0
commit 33518f1642
19 changed files with 1312 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gen/go/v1/config.pb.go
+102 -18
View File
@@ -260,6 +260,9 @@ const (
Hook_CONDITION_FORGET_START Hook_Condition = 300 // forget started.
Hook_CONDITION_FORGET_ERROR Hook_Condition = 301 // forget failed.
Hook_CONDITION_FORGET_SUCCESS Hook_Condition = 302 // forget succeeded.
// any operation conditions
Hook_CONDITION_ANY_START Hook_Condition = 400 // before any operation starts.
Hook_CONDITION_ANY_END Hook_Condition = 401 // after any operation ends (success or fail).
)
// Enum value maps for Hook_Condition.
@@ -282,6 +285,8 @@ var (
300: "CONDITION_FORGET_START",
301: "CONDITION_FORGET_ERROR",
302: "CONDITION_FORGET_SUCCESS",
400: "CONDITION_ANY_START",
401: "CONDITION_ANY_END",
}
Hook_Condition_value = map[string]int32{
"CONDITION_UNKNOWN": 0,
@@ -301,6 +306,8 @@ var (
"CONDITION_FORGET_START": 300,
"CONDITION_FORGET_ERROR": 301,
"CONDITION_FORGET_SUCCESS": 302,
"CONDITION_ANY_START": 400,
"CONDITION_ANY_END": 401,
}
)
@@ -1302,6 +1309,7 @@ type Hook struct {
// *Hook_ActionShoutrrr
// *Hook_ActionHealthchecks
// *Hook_ActionTelegram
// *Hook_ActionSyncLock
Action isHook_Action `protobuf_oneof:"action"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
@@ -1430,6 +1438,15 @@ func (x *Hook) GetActionTelegram() *Hook_Telegram {
return nil
}
func (x *Hook) GetActionSyncLock() *Hook_SyncLock {
if x != nil {
if x, ok := x.Action.(*Hook_ActionSyncLock); ok {
return x.ActionSyncLock
}
}
return nil
}
type isHook_Action interface {
isHook_Action()
}
@@ -1466,6 +1483,10 @@ type Hook_ActionTelegram struct {
ActionTelegram *Hook_Telegram `protobuf:"bytes,107,opt,name=action_telegram,json=actionTelegram,proto3,oneof"`
}
type Hook_ActionSyncLock struct {
ActionSyncLock *Hook_SyncLock `protobuf:"bytes,109,opt,name=action_sync_lock,json=actionSyncLock,proto3,oneof"`
}
func (*Hook_ActionCommand) isHook_Action() {}
func (*Hook_ActionWebhook) isHook_Action() {}
@@ -1482,6 +1503,8 @@ func (*Hook_ActionHealthchecks) isHook_Action() {}
func (*Hook_ActionTelegram) isHook_Action() {}
func (*Hook_ActionSyncLock) isHook_Action() {}
type Auth struct {
state protoimpl.MessageState `protogen:"open.v1"`
Disabled bool `protobuf:"varint,1,opt,name=disabled,proto3" json:"disabled,omitempty"` // disable authentication.
@@ -2364,6 +2387,58 @@ func (x *Hook_Telegram) GetTemplate() string {
return ""
}
type Hook_SyncLock struct {
state protoimpl.MessageState `protogen:"open.v1"`
TargetInstanceId string `protobuf:"bytes,1,opt,name=target_instance_id,json=targetInstanceId,proto3" json:"target_instance_id,omitempty"` // the instance ID of the peer to acquire the lock on.
LockKey string `protobuf:"bytes,2,opt,name=lock_key,json=lockKey,proto3" json:"lock_key,omitempty"` // the lock key, typically the repo ID.
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Hook_SyncLock) Reset() {
*x = Hook_SyncLock{}
mi := &file_v1_config_proto_msgTypes[24]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *Hook_SyncLock) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Hook_SyncLock) ProtoMessage() {}
func (x *Hook_SyncLock) ProtoReflect() protoreflect.Message {
mi := &file_v1_config_proto_msgTypes[24]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Hook_SyncLock.ProtoReflect.Descriptor instead.
func (*Hook_SyncLock) Descriptor() ([]byte, []int) {
return file_v1_config_proto_rawDescGZIP(), []int{9, 8}
}
func (x *Hook_SyncLock) GetTargetInstanceId() string {
if x != nil {
return x.TargetInstanceId
}
return ""
}
func (x *Hook_SyncLock) GetLockKey() string {
if x != nil {
return x.LockKey
}
return ""
}
var File_v1_config_proto protoreflect.FileDescriptor
const file_v1_config_proto_rawDesc = "" +
@@ -2484,7 +2559,7 @@ const file_v1_config_proto_rawDesc = "" +
"\tCLOCK_UTC\x10\x02\x12\x17\n" +
"\x13CLOCK_LAST_RUN_TIME\x10\x03B\n" +
"\n" +
"\bschedule\"\xe1\x0f\n" +
"\bschedule\"\xa7\x11\n" +
"\x04Hook\x122\n" +
"\n" +
"conditions\x18\x01 \x03(\x0e2\x12.v1.Hook.ConditionR\n" +
@@ -2497,7 +2572,8 @@ const file_v1_config_proto_rawDesc = "" +
"\faction_slack\x18h \x01(\v2\x0e.v1.Hook.SlackH\x00R\vactionSlack\x12<\n" +
"\x0faction_shoutrrr\x18i \x01(\v2\x11.v1.Hook.ShoutrrrH\x00R\x0eactionShoutrrr\x12H\n" +
"\x13action_healthchecks\x18j \x01(\v2\x15.v1.Hook.HealthchecksH\x00R\x12actionHealthchecks\x12<\n" +
"\x0faction_telegram\x18k \x01(\v2\x11.v1.Hook.TelegramH\x00R\x0eactionTelegram\x1a#\n" +
"\x0faction_telegram\x18k \x01(\v2\x11.v1.Hook.TelegramH\x00R\x0eactionTelegram\x12=\n" +
"\x10action_sync_lock\x18m \x01(\v2\x11.v1.Hook.SyncLockH\x00R\x0eactionSyncLock\x1a#\n" +
"\aCommand\x12\x18\n" +
"\acommand\x18\x01 \x01(\tR\acommand\x1a\xa1\x01\n" +
"\aWebhook\x12\x1f\n" +
@@ -2533,7 +2609,10 @@ const file_v1_config_proto_rawDesc = "" +
"\bTelegram\x12\x1b\n" +
"\tbot_token\x18\x01 \x01(\tR\bbotToken\x12\x17\n" +
"\achat_id\x18\x02 \x01(\tR\x06chatId\x12\x1a\n" +
"\btemplate\x18\x03 \x01(\tR\btemplate\"\xf5\x03\n" +
"\btemplate\x18\x03 \x01(\tR\btemplate\x1aS\n" +
"\bSyncLock\x12,\n" +
"\x12target_instance_id\x18\x01 \x01(\tR\x10targetInstanceId\x12\x19\n" +
"\block_key\x18\x02 \x01(\tR\alockKey\"\xa7\x04\n" +
"\tCondition\x12\x15\n" +
"\x11CONDITION_UNKNOWN\x10\x00\x12\x17\n" +
"\x13CONDITION_ANY_ERROR\x10\x01\x12\x1c\n" +
@@ -2551,7 +2630,9 @@ const file_v1_config_proto_rawDesc = "" +
"\x17CONDITION_CHECK_SUCCESS\x10\xca\x01\x12\x1b\n" +
"\x16CONDITION_FORGET_START\x10\xac\x02\x12\x1b\n" +
"\x16CONDITION_FORGET_ERROR\x10\xad\x02\x12\x1d\n" +
"\x18CONDITION_FORGET_SUCCESS\x10\xae\x02\"\xa9\x01\n" +
"\x18CONDITION_FORGET_SUCCESS\x10\xae\x02\x12\x18\n" +
"\x13CONDITION_ANY_START\x10\x90\x03\x12\x16\n" +
"\x11CONDITION_ANY_END\x10\x91\x03\"\xa9\x01\n" +
"\aOnError\x12\x13\n" +
"\x0fON_ERROR_IGNORE\x10\x00\x12\x13\n" +
"\x0fON_ERROR_CANCEL\x10\x01\x12\x12\n" +
@@ -2582,7 +2663,7 @@ func file_v1_config_proto_rawDescGZIP() []byte {
}
var file_v1_config_proto_enumTypes = make([]protoimpl.EnumInfo, 7)
var file_v1_config_proto_msgTypes = make([]protoimpl.MessageInfo, 24)
var file_v1_config_proto_msgTypes = make([]protoimpl.MessageInfo, 25)
var file_v1_config_proto_goTypes = []any{
(Multihost_Permission_Type)(0), // 0: v1.Multihost.Permission.Type
(CommandPrefix_IONiceLevel)(0), // 1: v1.CommandPrefix.IONiceLevel
@@ -2615,14 +2696,15 @@ var file_v1_config_proto_goTypes = []any{
(*Hook_Shoutrrr)(nil), // 28: v1.Hook.Shoutrrr
(*Hook_Healthchecks)(nil), // 29: v1.Hook.Healthchecks
(*Hook_Telegram)(nil), // 30: v1.Hook.Telegram
(*PrivateKey)(nil), // 31: v1.PrivateKey
(*Hook_SyncLock)(nil), // 31: v1.Hook.SyncLock
(*PrivateKey)(nil), // 32: v1.PrivateKey
}
var file_v1_config_proto_depIdxs = []int32{
9, // 0: v1.Config.repos:type_name -> v1.Repo
10, // 1: v1.Config.plans:type_name -> v1.Plan
17, // 2: v1.Config.auth:type_name -> v1.Auth
8, // 3: v1.Config.multihost:type_name -> v1.Multihost
31, // 4: v1.Multihost.identity:type_name -> v1.PrivateKey
32, // 4: v1.Multihost.identity:type_name -> v1.PrivateKey
19, // 5: v1.Multihost.known_hosts:type_name -> v1.Multihost.Peer
19, // 6: v1.Multihost.authorized_clients:type_name -> v1.Multihost.Peer
20, // 7: v1.Multihost.pairing_tokens:type_name -> v1.Multihost.PairingToken
@@ -2649,16 +2731,17 @@ var file_v1_config_proto_depIdxs = []int32{
28, // 28: v1.Hook.action_shoutrrr:type_name -> v1.Hook.Shoutrrr
29, // 29: v1.Hook.action_healthchecks:type_name -> v1.Hook.Healthchecks
30, // 30: v1.Hook.action_telegram:type_name -> v1.Hook.Telegram
18, // 31: v1.Auth.users:type_name -> v1.User
21, // 32: v1.Multihost.Peer.permissions:type_name -> v1.Multihost.Permission
21, // 33: v1.Multihost.PairingToken.permissions:type_name -> v1.Multihost.Permission
0, // 34: v1.Multihost.Permission.type:type_name -> v1.Multihost.Permission.Type
6, // 35: v1.Hook.Webhook.method:type_name -> v1.Hook.Webhook.Method
36, // [36:36] is the sub-list for method output_type
36, // [36:36] is the sub-list for method input_type
36, // [36:36] is the sub-list for extension type_name
36, // [36:36] is the sub-list for extension extendee
0, // [0:36] is the sub-list for field type_name
31, // 31: v1.Hook.action_sync_lock:type_name -> v1.Hook.SyncLock
18, // 32: v1.Auth.users:type_name -> v1.User
21, // 33: v1.Multihost.Peer.permissions:type_name -> v1.Multihost.Permission
21, // 34: v1.Multihost.PairingToken.permissions:type_name -> v1.Multihost.Permission
0, // 35: v1.Multihost.Permission.type:type_name -> v1.Multihost.Permission.Type
6, // 36: v1.Hook.Webhook.method:type_name -> v1.Hook.Webhook.Method
37, // [37:37] is the sub-list for method output_type
37, // [37:37] is the sub-list for method input_type
37, // [37:37] is the sub-list for extension type_name
37, // [37:37] is the sub-list for extension extendee
0, // [0:37] is the sub-list for field type_name
}
func init() { file_v1_config_proto_init() }
@@ -2691,6 +2774,7 @@ func file_v1_config_proto_init() {
(*Hook_ActionShoutrrr)(nil),
(*Hook_ActionHealthchecks)(nil),
(*Hook_ActionTelegram)(nil),
(*Hook_ActionSyncLock)(nil),
}
file_v1_config_proto_msgTypes[11].OneofWrappers = []any{
(*User_PasswordBcrypt)(nil),
@@ -2701,7 +2785,7 @@ func file_v1_config_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_v1_config_proto_rawDesc), len(file_v1_config_proto_rawDesc)),
NumEnums: 7,
NumMessages: 24,
NumMessages: 25,
NumExtensions: 0,
NumServices: 0,
},
gen/go/v1sync/syncservice.pb.go
+341 -40
View File
@@ -917,6 +917,10 @@ type SyncStreamItem struct {
// *SyncStreamItem_ReceiveResources
// *SyncStreamItem_RequestLog
// *SyncStreamItem_ReceiveLogData
// *SyncStreamItem_AcquireLock
// *SyncStreamItem_AcquireLockResponse
// *SyncStreamItem_ReleaseLock
// *SyncStreamItem_RefreshLock
// *SyncStreamItem_Throttle
Action isSyncStreamItem_Action `protobuf_oneof:"action"`
unknownFields protoimpl.UnknownFields
@@ -1068,6 +1072,42 @@ func (x *SyncStreamItem) GetReceiveLogData() *SyncStreamItem_SyncActionReceiveLo
return nil
}
func (x *SyncStreamItem) GetAcquireLock() *SyncStreamItem_SyncActionAcquireLock {
if x != nil {
if x, ok := x.Action.(*SyncStreamItem_AcquireLock); ok {
return x.AcquireLock
}
}
return nil
}
func (x *SyncStreamItem) GetAcquireLockResponse() *SyncStreamItem_SyncActionAcquireLockResponse {
if x != nil {
if x, ok := x.Action.(*SyncStreamItem_AcquireLockResponse); ok {
return x.AcquireLockResponse
}
}
return nil
}
func (x *SyncStreamItem) GetReleaseLock() *SyncStreamItem_SyncActionReleaseLock {
if x != nil {
if x, ok := x.Action.(*SyncStreamItem_ReleaseLock); ok {
return x.ReleaseLock
}
}
return nil
}
func (x *SyncStreamItem) GetRefreshLock() *SyncStreamItem_SyncActionRefreshLock {
if x != nil {
if x, ok := x.Action.(*SyncStreamItem_RefreshLock); ok {
return x.RefreshLock
}
}
return nil
}
func (x *SyncStreamItem) GetThrottle() *SyncStreamItem_SyncActionThrottle {
if x != nil {
if x, ok := x.Action.(*SyncStreamItem_Throttle); ok {
@@ -1129,6 +1169,22 @@ type SyncStreamItem_ReceiveLogData struct {
ReceiveLogData *SyncStreamItem_SyncActionReceiveLogData `protobuf:"bytes,31,opt,name=receive_log_data,json=receiveLogData,proto3,oneof"`
}
type SyncStreamItem_AcquireLock struct {
AcquireLock *SyncStreamItem_SyncActionAcquireLock `protobuf:"bytes,40,opt,name=acquire_lock,json=acquireLock,proto3,oneof"`
}
type SyncStreamItem_AcquireLockResponse struct {
AcquireLockResponse *SyncStreamItem_SyncActionAcquireLockResponse `protobuf:"bytes,41,opt,name=acquire_lock_response,json=acquireLockResponse,proto3,oneof"`
}
type SyncStreamItem_ReleaseLock struct {
ReleaseLock *SyncStreamItem_SyncActionReleaseLock `protobuf:"bytes,42,opt,name=release_lock,json=releaseLock,proto3,oneof"`
}
type SyncStreamItem_RefreshLock struct {
RefreshLock *SyncStreamItem_SyncActionRefreshLock `protobuf:"bytes,43,opt,name=refresh_lock,json=refreshLock,proto3,oneof"`
}
type SyncStreamItem_Throttle struct {
Throttle *SyncStreamItem_SyncActionThrottle `protobuf:"bytes,1000,opt,name=throttle,proto3,oneof"`
}
@@ -1157,6 +1213,14 @@ func (*SyncStreamItem_RequestLog) isSyncStreamItem_Action() {}
func (*SyncStreamItem_ReceiveLogData) isSyncStreamItem_Action() {}
func (*SyncStreamItem_AcquireLock) isSyncStreamItem_Action() {}
func (*SyncStreamItem_AcquireLockResponse) isSyncStreamItem_Action() {}
func (*SyncStreamItem_ReleaseLock) isSyncStreamItem_Action() {}
func (*SyncStreamItem_RefreshLock) isSyncStreamItem_Action() {}
func (*SyncStreamItem_Throttle) isSyncStreamItem_Action() {}
type SyncStreamItem_SyncActionHandshake struct {
@@ -1861,6 +1925,215 @@ func (x *SyncStreamItem_SyncEstablishSharedSecret) GetEcdsaPub() string {
return ""
}
// Lock protocol messages for coordinating repo access between peers.
type SyncStreamItem_SyncActionAcquireLock struct {
state protoimpl.MessageState `protogen:"open.v1"`
LockKey string `protobuf:"bytes,1,opt,name=lock_key,json=lockKey,proto3" json:"lock_key,omitempty"` // the lock key, typically a repo ID.
HolderId string `protobuf:"bytes,2,opt,name=holder_id,json=holderId,proto3" json:"holder_id,omitempty"` // the instance ID of the lock requester.
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SyncStreamItem_SyncActionAcquireLock) Reset() {
*x = SyncStreamItem_SyncActionAcquireLock{}
mi := &file_v1sync_syncservice_proto_msgTypes[28]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SyncStreamItem_SyncActionAcquireLock) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SyncStreamItem_SyncActionAcquireLock) ProtoMessage() {}
func (x *SyncStreamItem_SyncActionAcquireLock) ProtoReflect() protoreflect.Message {
mi := &file_v1sync_syncservice_proto_msgTypes[28]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use SyncStreamItem_SyncActionAcquireLock.ProtoReflect.Descriptor instead.
func (*SyncStreamItem_SyncActionAcquireLock) Descriptor() ([]byte, []int) {
return file_v1sync_syncservice_proto_rawDescGZIP(), []int{13, 14}
}
func (x *SyncStreamItem_SyncActionAcquireLock) GetLockKey() string {
if x != nil {
return x.LockKey
}
return ""
}
func (x *SyncStreamItem_SyncActionAcquireLock) GetHolderId() string {
if x != nil {
return x.HolderId
}
return ""
}
type SyncStreamItem_SyncActionAcquireLockResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Acquired bool `protobuf:"varint,1,opt,name=acquired,proto3" json:"acquired,omitempty"` // whether the lock was acquired.
LockKey string `protobuf:"bytes,2,opt,name=lock_key,json=lockKey,proto3" json:"lock_key,omitempty"` // the lock key that was requested.
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SyncStreamItem_SyncActionAcquireLockResponse) Reset() {
*x = SyncStreamItem_SyncActionAcquireLockResponse{}
mi := &file_v1sync_syncservice_proto_msgTypes[29]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SyncStreamItem_SyncActionAcquireLockResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SyncStreamItem_SyncActionAcquireLockResponse) ProtoMessage() {}
func (x *SyncStreamItem_SyncActionAcquireLockResponse) ProtoReflect() protoreflect.Message {
mi := &file_v1sync_syncservice_proto_msgTypes[29]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use SyncStreamItem_SyncActionAcquireLockResponse.ProtoReflect.Descriptor instead.
func (*SyncStreamItem_SyncActionAcquireLockResponse) Descriptor() ([]byte, []int) {
return file_v1sync_syncservice_proto_rawDescGZIP(), []int{13, 15}
}
func (x *SyncStreamItem_SyncActionAcquireLockResponse) GetAcquired() bool {
if x != nil {
return x.Acquired
}
return false
}
func (x *SyncStreamItem_SyncActionAcquireLockResponse) GetLockKey() string {
if x != nil {
return x.LockKey
}
return ""
}
type SyncStreamItem_SyncActionReleaseLock struct {
state protoimpl.MessageState `protogen:"open.v1"`
LockKey string `protobuf:"bytes,1,opt,name=lock_key,json=lockKey,proto3" json:"lock_key,omitempty"` // the lock key to release.
HolderId string `protobuf:"bytes,2,opt,name=holder_id,json=holderId,proto3" json:"holder_id,omitempty"` // the instance ID of the lock holder.
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SyncStreamItem_SyncActionReleaseLock) Reset() {
*x = SyncStreamItem_SyncActionReleaseLock{}
mi := &file_v1sync_syncservice_proto_msgTypes[30]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SyncStreamItem_SyncActionReleaseLock) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SyncStreamItem_SyncActionReleaseLock) ProtoMessage() {}
func (x *SyncStreamItem_SyncActionReleaseLock) ProtoReflect() protoreflect.Message {
mi := &file_v1sync_syncservice_proto_msgTypes[30]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use SyncStreamItem_SyncActionReleaseLock.ProtoReflect.Descriptor instead.
func (*SyncStreamItem_SyncActionReleaseLock) Descriptor() ([]byte, []int) {
return file_v1sync_syncservice_proto_rawDescGZIP(), []int{13, 16}
}
func (x *SyncStreamItem_SyncActionReleaseLock) GetLockKey() string {
if x != nil {
return x.LockKey
}
return ""
}
func (x *SyncStreamItem_SyncActionReleaseLock) GetHolderId() string {
if x != nil {
return x.HolderId
}
return ""
}
type SyncStreamItem_SyncActionRefreshLock struct {
state protoimpl.MessageState `protogen:"open.v1"`
LockKey string `protobuf:"bytes,1,opt,name=lock_key,json=lockKey,proto3" json:"lock_key,omitempty"` // the lock key to refresh.
HolderId string `protobuf:"bytes,2,opt,name=holder_id,json=holderId,proto3" json:"holder_id,omitempty"` // the instance ID of the lock holder.
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SyncStreamItem_SyncActionRefreshLock) Reset() {
*x = SyncStreamItem_SyncActionRefreshLock{}
mi := &file_v1sync_syncservice_proto_msgTypes[31]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SyncStreamItem_SyncActionRefreshLock) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SyncStreamItem_SyncActionRefreshLock) ProtoMessage() {}
func (x *SyncStreamItem_SyncActionRefreshLock) ProtoReflect() protoreflect.Message {
mi := &file_v1sync_syncservice_proto_msgTypes[31]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use SyncStreamItem_SyncActionRefreshLock.ProtoReflect.Descriptor instead.
func (*SyncStreamItem_SyncActionRefreshLock) Descriptor() ([]byte, []int) {
return file_v1sync_syncservice_proto_rawDescGZIP(), []int{13, 17}
}
func (x *SyncStreamItem_SyncActionRefreshLock) GetLockKey() string {
if x != nil {
return x.LockKey
}
return ""
}
func (x *SyncStreamItem_SyncActionRefreshLock) GetHolderId() string {
if x != nil {
return x.HolderId
}
return ""
}
var File_v1sync_syncservice_proto protoreflect.FileDescriptor
const file_v1sync_syncservice_proto_rawDesc = "" +
@@ -1922,7 +2195,7 @@ const file_v1sync_syncservice_proto_rawDesc = "" +
"\n" +
"public_key\x18\x01 \x01(\v2\r.v1.PublicKeyR\tpublicKey\x122\n" +
"\vinstance_id\x18\x02 \x01(\v2\x11.v1.SignedMessageR\n" +
"instanceId\"\xff\x13\n" +
"instanceId\"\xaf\x19\n" +
"\x0eSyncStreamItem\x12:\n" +
"\x0esigned_message\x18\x01 \x01(\v2\x11.v1.SignedMessageH\x00R\rsignedMessage\x12J\n" +
"\thandshake\x18\x03 \x01(\v2*.v1sync.SyncStreamItem.SyncActionHandshakeH\x00R\thandshake\x12J\n" +
@@ -1937,7 +2210,11 @@ const file_v1sync_syncservice_proto_rawDesc = "" +
"\x11receive_resources\x18\x1a \x01(\v21.v1sync.SyncStreamItem.SyncActionReceiveResourcesH\x00R\x10receiveResources\x12N\n" +
"\vrequest_log\x18\x1e \x01(\v2+.v1sync.SyncStreamItem.SyncActionRequestLogH\x00R\n" +
"requestLog\x12[\n" +
"\x10receive_log_data\x18\x1f \x01(\v2/.v1sync.SyncStreamItem.SyncActionReceiveLogDataH\x00R\x0ereceiveLogData\x12H\n" +
"\x10receive_log_data\x18\x1f \x01(\v2/.v1sync.SyncStreamItem.SyncActionReceiveLogDataH\x00R\x0ereceiveLogData\x12Q\n" +
"\facquire_lock\x18( \x01(\v2,.v1sync.SyncStreamItem.SyncActionAcquireLockH\x00R\vacquireLock\x12j\n" +
"\x15acquire_lock_response\x18) \x01(\v24.v1sync.SyncStreamItem.SyncActionAcquireLockResponseH\x00R\x13acquireLockResponse\x12Q\n" +
"\frelease_lock\x18* \x01(\v2,.v1sync.SyncStreamItem.SyncActionReleaseLockH\x00R\vreleaseLock\x12Q\n" +
"\frefresh_lock\x18+ \x01(\v2,.v1sync.SyncStreamItem.SyncActionRefreshLockH\x00R\vrefreshLock\x12H\n" +
"\bthrottle\x18\xe8\a \x01(\v2).v1sync.SyncStreamItem.SyncActionThrottleH\x00R\bthrottle\x1a\xc9\x01\n" +
"\x13SyncActionHandshake\x12)\n" +
"\x10protocol_version\x18\x01 \x01(\x03R\x0fprotocolVersion\x12,\n" +
@@ -1979,7 +2256,19 @@ const file_v1sync_syncservice_proto_rawDesc = "" +
"\x12SyncActionThrottle\x12\x19\n" +
"\bdelay_ms\x18\x01 \x01(\x03R\adelayMs\x1a8\n" +
"\x19SyncEstablishSharedSecret\x12\x1b\n" +
"\tecdsa_pub\x18\x02 \x01(\tR\becdsaPub\"\xb4\x01\n" +
"\tecdsa_pub\x18\x02 \x01(\tR\becdsaPub\x1aO\n" +
"\x15SyncActionAcquireLock\x12\x19\n" +
"\block_key\x18\x01 \x01(\tR\alockKey\x12\x1b\n" +
"\tholder_id\x18\x02 \x01(\tR\bholderId\x1aV\n" +
"\x1dSyncActionAcquireLockResponse\x12\x1a\n" +
"\bacquired\x18\x01 \x01(\bR\bacquired\x12\x19\n" +
"\block_key\x18\x02 \x01(\tR\alockKey\x1aO\n" +
"\x15SyncActionReleaseLock\x12\x19\n" +
"\block_key\x18\x01 \x01(\tR\alockKey\x12\x1b\n" +
"\tholder_id\x18\x02 \x01(\tR\bholderId\x1aO\n" +
"\x15SyncActionRefreshLock\x12\x19\n" +
"\block_key\x18\x01 \x01(\tR\alockKey\x12\x1b\n" +
"\tholder_id\x18\x02 \x01(\tR\bholderId\"\xb4\x01\n" +
"\x13RepoConnectionState\x12\x1c\n" +
"\x18CONNECTION_STATE_UNKNOWN\x10\x00\x12\x1c\n" +
"\x18CONNECTION_STATE_PENDING\x10\x01\x12\x1e\n" +
@@ -2016,7 +2305,7 @@ func file_v1sync_syncservice_proto_rawDescGZIP() []byte {
}
var file_v1sync_syncservice_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
var file_v1sync_syncservice_proto_msgTypes = make([]protoimpl.MessageInfo, 28)
var file_v1sync_syncservice_proto_msgTypes = make([]protoimpl.MessageInfo, 32)
var file_v1sync_syncservice_proto_goTypes = []any{
(ConnectionState)(0), // 0: v1sync.ConnectionState
(SyncStreamItem_RepoConnectionState)(0), // 1: v1sync.SyncStreamItem.RepoConnectionState
@@ -2048,29 +2337,33 @@ var file_v1sync_syncservice_proto_goTypes = []any{
(*SyncStreamItem_SyncActionReceiveLogData)(nil), // 27: v1sync.SyncStreamItem.SyncActionReceiveLogData
(*SyncStreamItem_SyncActionThrottle)(nil), // 28: v1sync.SyncStreamItem.SyncActionThrottle
(*SyncStreamItem_SyncEstablishSharedSecret)(nil), // 29: v1sync.SyncStreamItem.SyncEstablishSharedSecret
(*v1.SignedMessage)(nil), // 30: v1.SignedMessage
(*v1.Plan)(nil), // 31: v1.Plan
(*v1.Repo)(nil), // 32: v1.Repo
(*v1.PublicKey)(nil), // 33: v1.PublicKey
(*v1.OperationEvent)(nil), // 34: v1.OperationEvent
(*SyncStreamItem_SyncActionAcquireLock)(nil), // 30: v1sync.SyncStreamItem.SyncActionAcquireLock
(*SyncStreamItem_SyncActionAcquireLockResponse)(nil), // 31: v1sync.SyncStreamItem.SyncActionAcquireLockResponse
(*SyncStreamItem_SyncActionReleaseLock)(nil), // 32: v1sync.SyncStreamItem.SyncActionReleaseLock
(*SyncStreamItem_SyncActionRefreshLock)(nil), // 33: v1sync.SyncStreamItem.SyncActionRefreshLock
(*v1.SignedMessage)(nil), // 34: v1.SignedMessage
(*v1.Plan)(nil), // 35: v1.Plan
(*v1.Repo)(nil), // 36: v1.Repo
(*v1.PublicKey)(nil), // 37: v1.PublicKey
(*v1.OperationEvent)(nil), // 38: v1.OperationEvent
}
var file_v1sync_syncservice_proto_depIdxs = []int32{
0, // 0: v1sync.PeerState.state:type_name -> v1sync.ConnectionState
9, // 1: v1sync.PeerState.known_plans:type_name -> v1sync.PlanMetadata
8, // 2: v1sync.PeerState.known_repos:type_name -> v1sync.RepoMetadata
13, // 3: v1sync.PeerState.remote_config:type_name -> v1sync.RemoteConfig
30, // 4: v1sync.AuthenticateRequest.instance_id:type_name -> v1.SignedMessage
34, // 4: v1sync.AuthenticateRequest.instance_id:type_name -> v1.SignedMessage
9, // 5: v1sync.SetAvailableResourcesRequest.repos:type_name -> v1sync.PlanMetadata
8, // 6: v1sync.SetAvailableResourcesRequest.plans:type_name -> v1sync.RepoMetadata
31, // 7: v1sync.SetConfigRequest.plans:type_name -> v1.Plan
32, // 8: v1sync.SetConfigRequest.repos:type_name -> v1.Repo
32, // 9: v1sync.SetRemoteClientConfigRequest.repos:type_name -> v1.Repo
31, // 10: v1sync.SetRemoteClientConfigRequest.plans:type_name -> v1.Plan
32, // 11: v1sync.RemoteConfig.repos:type_name -> v1.Repo
31, // 12: v1sync.RemoteConfig.plans:type_name -> v1.Plan
33, // 13: v1sync.AuthorizationToken.public_key:type_name -> v1.PublicKey
30, // 14: v1sync.AuthorizationToken.instance_id:type_name -> v1.SignedMessage
30, // 15: v1sync.SyncStreamItem.signed_message:type_name -> v1.SignedMessage
35, // 7: v1sync.SetConfigRequest.plans:type_name -> v1.Plan
36, // 8: v1sync.SetConfigRequest.repos:type_name -> v1.Repo
36, // 9: v1sync.SetRemoteClientConfigRequest.repos:type_name -> v1.Repo
35, // 10: v1sync.SetRemoteClientConfigRequest.plans:type_name -> v1.Plan
36, // 11: v1sync.RemoteConfig.repos:type_name -> v1.Repo
35, // 12: v1sync.RemoteConfig.plans:type_name -> v1.Plan
37, // 13: v1sync.AuthorizationToken.public_key:type_name -> v1.PublicKey
34, // 14: v1sync.AuthorizationToken.instance_id:type_name -> v1.SignedMessage
34, // 15: v1sync.SyncStreamItem.signed_message:type_name -> v1.SignedMessage
16, // 16: v1sync.SyncStreamItem.handshake:type_name -> v1sync.SyncStreamItem.SyncActionHandshake
17, // 17: v1sync.SyncStreamItem.heartbeat:type_name -> v1sync.SyncStreamItem.SyncActionHeartbeat
23, // 18: v1sync.SyncStreamItem.operation_manifest:type_name -> v1sync.SyncStreamItem.SyncActionOperationManifest
@@ -2082,26 +2375,30 @@ var file_v1sync_syncservice_proto_depIdxs = []int32{
21, // 24: v1sync.SyncStreamItem.receive_resources:type_name -> v1sync.SyncStreamItem.SyncActionReceiveResources
26, // 25: v1sync.SyncStreamItem.request_log:type_name -> v1sync.SyncStreamItem.SyncActionRequestLog
27, // 26: v1sync.SyncStreamItem.receive_log_data:type_name -> v1sync.SyncStreamItem.SyncActionReceiveLogData
28, // 27: v1sync.SyncStreamItem.throttle:type_name -> v1sync.SyncStreamItem.SyncActionThrottle
33, // 28: v1sync.SyncStreamItem.SyncActionHandshake.public_key:type_name -> v1.PublicKey
30, // 29: v1sync.SyncStreamItem.SyncActionHandshake.instance_id:type_name -> v1.SignedMessage
13, // 30: v1sync.SyncStreamItem.SyncActionReceiveConfig.config:type_name -> v1sync.RemoteConfig
32, // 31: v1sync.SyncStreamItem.SyncActionSetConfig.repos:type_name -> v1.Repo
31, // 32: v1sync.SyncStreamItem.SyncActionSetConfig.plans:type_name -> v1.Plan
8, // 33: v1sync.SyncStreamItem.SyncActionReceiveResources.repos:type_name -> v1sync.RepoMetadata
9, // 34: v1sync.SyncStreamItem.SyncActionReceiveResources.plans:type_name -> v1sync.PlanMetadata
34, // 35: v1sync.SyncStreamItem.SyncActionReceiveOperations.event:type_name -> v1.OperationEvent
15, // 36: v1sync.BackrestSyncService.Sync:input_type -> v1sync.SyncStreamItem
2, // 37: v1sync.BackrestSyncStateService.GetPeerSyncStatesStream:input_type -> v1sync.SyncStateStreamRequest
11, // 38: v1sync.BackrestSyncStateService.SetRemoteClientConfig:input_type -> v1sync.SetRemoteClientConfigRequest
15, // 39: v1sync.BackrestSyncService.Sync:output_type -> v1sync.SyncStreamItem
3, // 40: v1sync.BackrestSyncStateService.GetPeerSyncStatesStream:output_type -> v1sync.PeerState
12, // 41: v1sync.BackrestSyncStateService.SetRemoteClientConfig:output_type -> v1sync.SetRemoteClientConfigResponse
39, // [39:42] is the sub-list for method output_type
36, // [36:39] is the sub-list for method input_type
36, // [36:36] is the sub-list for extension type_name
36, // [36:36] is the sub-list for extension extendee
0, // [0:36] is the sub-list for field type_name
30, // 27: v1sync.SyncStreamItem.acquire_lock:type_name -> v1sync.SyncStreamItem.SyncActionAcquireLock
31, // 28: v1sync.SyncStreamItem.acquire_lock_response:type_name -> v1sync.SyncStreamItem.SyncActionAcquireLockResponse
32, // 29: v1sync.SyncStreamItem.release_lock:type_name -> v1sync.SyncStreamItem.SyncActionReleaseLock
33, // 30: v1sync.SyncStreamItem.refresh_lock:type_name -> v1sync.SyncStreamItem.SyncActionRefreshLock
28, // 31: v1sync.SyncStreamItem.throttle:type_name -> v1sync.SyncStreamItem.SyncActionThrottle
37, // 32: v1sync.SyncStreamItem.SyncActionHandshake.public_key:type_name -> v1.PublicKey
34, // 33: v1sync.SyncStreamItem.SyncActionHandshake.instance_id:type_name -> v1.SignedMessage
13, // 34: v1sync.SyncStreamItem.SyncActionReceiveConfig.config:type_name -> v1sync.RemoteConfig
36, // 35: v1sync.SyncStreamItem.SyncActionSetConfig.repos:type_name -> v1.Repo
35, // 36: v1sync.SyncStreamItem.SyncActionSetConfig.plans:type_name -> v1.Plan
8, // 37: v1sync.SyncStreamItem.SyncActionReceiveResources.repos:type_name -> v1sync.RepoMetadata
9, // 38: v1sync.SyncStreamItem.SyncActionReceiveResources.plans:type_name -> v1sync.PlanMetadata
38, // 39: v1sync.SyncStreamItem.SyncActionReceiveOperations.event:type_name -> v1.OperationEvent
15, // 40: v1sync.BackrestSyncService.Sync:input_type -> v1sync.SyncStreamItem
2, // 41: v1sync.BackrestSyncStateService.GetPeerSyncStatesStream:input_type -> v1sync.SyncStateStreamRequest
11, // 42: v1sync.BackrestSyncStateService.SetRemoteClientConfig:input_type -> v1sync.SetRemoteClientConfigRequest
15, // 43: v1sync.BackrestSyncService.Sync:output_type -> v1sync.SyncStreamItem
3, // 44: v1sync.BackrestSyncStateService.GetPeerSyncStatesStream:output_type -> v1sync.PeerState
12, // 45: v1sync.BackrestSyncStateService.SetRemoteClientConfig:output_type -> v1sync.SetRemoteClientConfigResponse
43, // [43:46] is the sub-list for method output_type
40, // [40:43] is the sub-list for method input_type
40, // [40:40] is the sub-list for extension type_name
40, // [40:40] is the sub-list for extension extendee
0, // [0:40] is the sub-list for field type_name
}
func init() { file_v1sync_syncservice_proto_init() }
@@ -2122,6 +2419,10 @@ func file_v1sync_syncservice_proto_init() {
(*SyncStreamItem_ReceiveResources)(nil),
(*SyncStreamItem_RequestLog)(nil),
(*SyncStreamItem_ReceiveLogData)(nil),
(*SyncStreamItem_AcquireLock)(nil),
(*SyncStreamItem_AcquireLockResponse)(nil),
(*SyncStreamItem_ReleaseLock)(nil),
(*SyncStreamItem_RefreshLock)(nil),
(*SyncStreamItem_Throttle)(nil),
}
type x struct{}
@@ -2130,7 +2431,7 @@ func file_v1sync_syncservice_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_v1sync_syncservice_proto_rawDesc), len(file_v1sync_syncservice_proto_rawDesc)),
NumEnums: 2,
NumMessages: 28,
NumMessages: 32,
NumExtensions: 0,
NumServices: 2,
},
internal/api/syncapi/authmiddleware.go
+7 -6
View File
@@ -52,7 +52,7 @@ func newAuthHandler(config *config.ConfigManager, next http.Handler) http.Handle
}
rw.Header().Set(authTokenHeader, authHeaderValue)
peer, err := decodeAndVerifyAuthHeader(r, config.Instance, config.GetMultihost().GetAuthorizedClients())
peer, err := decodeAndVerifyAuthHeader(r.Header, config.GetMultihost().GetAuthorizedClients())
if err != nil {
http.Error(rw, fmt.Sprintf("unauthorized: %v", err), http.StatusUnauthorized)
return
@@ -116,7 +116,8 @@ func (c *authHeaderClient) Do(req *http.Request) (*http.Response, error) {
if resp.StatusCode != http.StatusOK {
return resp, fmt.Errorf("HTTP request failed with status %d: %s", resp.StatusCode, resp.Status)
}
peer, err := decodeAndVerifyAuthHeader(req, cfg.Instance, cfg.GetMultihost().GetAuthorizedClients())
// Verify the server's response header against our known hosts.
peer, err := decodeAndVerifyAuthHeader(resp.Header, cfg.GetMultihost().GetKnownHosts())
if err != nil {
return resp, fmt.Errorf("verify auth header: %w", err)
}
@@ -138,8 +139,8 @@ func newHTTPClientWithConfig(cfg *config.ConfigManager, delegate connect.HTTPCli
}, nil
}
func decodeAndVerifyAuthHeader(r *http.Request, localInstanceID string, peers []*v1.Multihost_Peer) (*v1.Multihost_Peer, error) {
authHeader := r.Header.Get(authTokenHeader)
func decodeAndVerifyAuthHeader(header http.Header, peers []*v1.Multihost_Peer) (*v1.Multihost_Peer, error) {
authHeader := header.Get(authTokenHeader)
if len(authHeader) == 0 {
return nil, errors.New("missing authorization header")
}
@@ -195,7 +196,7 @@ func createSignedMessage(payload []byte, identity *cryptoutil.PrivateKey) (*v1.S
timestampMillis := time.Now().UnixMilli()
payloadWithTimestamp := make([]byte, 0, len(payload)+8)
binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(timestampMillis))
payloadWithTimestamp = binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(timestampMillis))
payloadWithTimestamp = append(payloadWithTimestamp, payload...)
signature, err := identity.Sign(payloadWithTimestamp)
@@ -230,7 +231,7 @@ func verifySignedMessage(msg *v1.SignedMessage, publicKey *cryptoutil.PublicKey)
}
payloadWithTimestamp := make([]byte, 0, len(msg.GetPayload())+8)
binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(msg.GetTimestampMillis()))
payloadWithTimestamp = binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(msg.GetTimestampMillis()))
payloadWithTimestamp = append(payloadWithTimestamp, msg.GetPayload()...)
if err := publicKey.Verify(payloadWithTimestamp, msg.GetSignature()); err != nil {
internal/api/syncapi/cmdstreamutil.go
+11 -7
View File
@@ -27,7 +27,7 @@ type bidiSyncCommandStream struct {
func newBidiSyncCommandStream() *bidiSyncCommandStream {
return &bidiSyncCommandStream{
sendChan: make(chan *v1sync.SyncStreamItem, 64), // Buffered channel to allow sending items without blocking
sendChan: make(chan *v1sync.SyncStreamItem, 256), // Buffered channel to allow sending items without blocking
recvChan: make(chan *v1sync.SyncStreamItem, 1),
terminateWithErrChan: make(chan error, 1),
}
@@ -74,15 +74,19 @@ func (s *bidiSyncCommandStream) ConnectStream(ctx context.Context, stream syncCo
ctx, cancel := context.WithCancel(ctx)
defer cancel()
go func() {
for ctx.Err() == nil {
if val, err := stream.Receive(); err != nil {
defer close(s.recvChan)
for {
val, err := stream.Receive()
if err != nil {
s.SendErrorAndTerminate(NewSyncErrorDisconnected(fmt.Errorf("receiving item: %w", err)))
break
} else {
s.recvChan <- val
return
}
select {
case s.recvChan <- val:
case <-ctx.Done():
return
}
}
close(s.recvChan)
}()
for {
internal/api/syncapi/lockmanager.go
+69
View File
@@ -0,0 +1,69 @@
package syncapi
import (
"sync"
"time"
)
const lockExpiry = 30 * time.Second
type lockEntry struct {
holderID string
expiresAt time.Time
}
// LockManager provides an in-memory best-effort lock store for coordinating
// repo access between sync peers. Locks expire after 30 seconds if not refreshed.
type LockManager struct {
mu sync.Mutex
locks map[string]*lockEntry
}
func NewLockManager() *LockManager {
return &LockManager{
locks: make(map[string]*lockEntry),
}
}
// Acquire attempts to acquire a lock for the given key. Returns true if the
// lock was acquired (key was free, expired, or already held by the same holder).
func (lm *LockManager) Acquire(key, holderID string) bool {
lm.mu.Lock()
defer lm.mu.Unlock()
entry, exists := lm.locks[key]
if !exists || time.Now().After(entry.expiresAt) || entry.holderID == holderID {
lm.locks[key] = &lockEntry{
holderID: holderID,
expiresAt: time.Now().Add(lockExpiry),
}
return true
}
return false
}
// Release releases a lock if the holder matches.
func (lm *LockManager) Release(key, holderID string) bool {
lm.mu.Lock()
defer lm.mu.Unlock()
entry, exists := lm.locks[key]
if exists && entry.holderID == holderID {
delete(lm.locks, key)
return true
}
return false
}
// Refresh extends the expiry of a lock if the holder matches.
func (lm *LockManager) Refresh(key, holderID string) bool {
lm.mu.Lock()
defer lm.mu.Unlock()
entry, exists := lm.locks[key]
if exists && entry.holderID == holderID {
entry.expiresAt = time.Now().Add(lockExpiry)
return true
}
return false
}
internal/api/syncapi/syncclient.go
+133 -11
View File
@@ -35,6 +35,11 @@ type SyncClient struct {
l *zap.Logger
reconnectAttempts int
// Lock protocol support
streamMu sync.Mutex
activeStream *bidiSyncCommandStream // set while connected, nil otherwise
lockResponseCh chan *v1sync.SyncStreamItem_SyncActionAcquireLockResponse
}
func newInsecureClient() *http.Client {
@@ -73,6 +78,7 @@ func NewSyncClient(
client: client,
oplog: oplog,
l: zap.L().Named(fmt.Sprintf("syncclient for %q", peer.GetInstanceId())),
lockResponseCh: make(chan *v1sync.SyncStreamItem_SyncActionAcquireLockResponse, 1),
}
c.mgr.peerStateManager.SetPeerState(peer.Keyid, newPeerState(peer.InstanceId, peer.Keyid))
return c, nil
@@ -92,10 +98,15 @@ func (c *SyncClient) RunSync(ctx context.Context) {
c.syncConfigSnapshot,
c.oplog,
c.peer,
c,
)
cmdStream := newBidiSyncCommandStream()
c.streamMu.Lock()
c.activeStream = cmdStream
c.streamMu.Unlock()
c.l.Sugar().Infof("connecting to peer %q (%s) at %s", c.peer.InstanceId, c.peer.Keyid, c.peer.GetInstanceUrl())
var wg sync.WaitGroup
@@ -115,33 +126,45 @@ func (c *SyncClient) RunSync(ctx context.Context) {
cmdStream.SendErrorAndTerminate(err)
}()
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)
connectErr := cmdStream.ConnectStream(ctx, c.client.Sync(ctx))
c.streamMu.Lock()
c.activeStream = nil
c.streamMu.Unlock()
if connectErr != nil {
c.l.Sugar().Infof("lost stream connection to peer %q (%s): %v", c.peer.InstanceId, c.peer.Keyid, connectErr)
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) {
if errors.As(connectErr, &syncErr) {
state.ConnectionState = syncErr.State
state.ConnectionStateMessage = syncErr.Message.Error()
} else {
state.ConnectionState = v1sync.ConnectionState_CONNECTION_STATE_ERROR_INTERNAL
state.ConnectionStateMessage = err.Error()
state.ConnectionStateMessage = connectErr.Error()
}
c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, state)
} else {
c.reconnectAttempts = 0
}
wg.Wait()
// Reset reconnect backoff if the session lasted long enough to be considered a real success,
// rather than a handshake that failed immediately. Using reconnectDelay as the threshold means
// any session that ran at least one full retry window counts as stable.
if time.Since(lastConnect) >= c.reconnectDelay {
c.reconnectAttempts = 0
}
delay := c.reconnectDelay - time.Since(lastConnect)
if c.reconnectAttempts > 0 {
backoff := time.Duration(1<<min(c.reconnectAttempts, 5)) * c.reconnectDelay // 2^reconnectAttempts, max 32
delay += backoff
}
if delay < 0 {
delay = 0
}
c.l.Sugar().Infof("disconnected, will retry after %v (attempt %d)", delay, c.reconnectAttempts)
c.reconnectAttempts++
select {
@@ -152,6 +175,83 @@ func (c *SyncClient) RunSync(ctx context.Context) {
}
}
// AcquireLock sends a lock acquire request and waits for the response.
// Returns true if the lock was acquired, false otherwise.
// Returns an error if the client is not connected or the context expires.
func (c *SyncClient) AcquireLock(ctx context.Context, lockKey string) (bool, error) {
c.streamMu.Lock()
stream := c.activeStream
c.streamMu.Unlock()
if stream == nil {
return false, fmt.Errorf("not connected to peer %q", c.peer.GetInstanceId())
}
// Drain any stale responses
select {
case <-c.lockResponseCh:
default:
}
stream.Send(&v1sync.SyncStreamItem{
Action: &v1sync.SyncStreamItem_AcquireLock{
AcquireLock: &v1sync.SyncStreamItem_SyncActionAcquireLock{
LockKey: lockKey,
HolderId: c.localInstanceID,
},
},
})
select {
case resp := <-c.lockResponseCh:
return resp.GetAcquired(), nil
case <-ctx.Done():
return false, ctx.Err()
case <-time.After(10 * time.Second):
return false, fmt.Errorf("timeout waiting for lock response from peer %q", c.peer.GetInstanceId())
}
}
// ReleaseLock sends a lock release request to the peer.
func (c *SyncClient) ReleaseLock(lockKey string) {
c.streamMu.Lock()
stream := c.activeStream
c.streamMu.Unlock()
if stream == nil {
return
}
stream.Send(&v1sync.SyncStreamItem{
Action: &v1sync.SyncStreamItem_ReleaseLock{
ReleaseLock: &v1sync.SyncStreamItem_SyncActionReleaseLock{
LockKey: lockKey,
HolderId: c.localInstanceID,
},
},
})
}
// RefreshLock sends a lock refresh request to the peer.
func (c *SyncClient) RefreshLock(lockKey string) {
c.streamMu.Lock()
stream := c.activeStream
c.streamMu.Unlock()
if stream == nil {
return
}
stream.Send(&v1sync.SyncStreamItem{
Action: &v1sync.SyncStreamItem_RefreshLock{
RefreshLock: &v1sync.SyncStreamItem_SyncActionRefreshLock{
LockKey: lockKey,
HolderId: c.localInstanceID,
},
},
})
}
// syncSessionHandlerClient is a syncSessionHandler implementation for clients.
type syncSessionHandlerClient struct {
unimplementedSyncSessionHandler
@@ -167,6 +267,10 @@ type syncSessionHandlerClient struct {
canForwardReposSet map[string]struct{}
canForwardPlansSet map[string]struct{}
syncClient *SyncClient // back-reference for forwarding lock responses
oplogSubscription *oplog.Subscription // set while subscribed; unsubscribed in OnConnectionDisconnected.
}
func newSyncHandlerClient(
@@ -175,6 +279,7 @@ func newSyncHandlerClient(
snapshot syncConfigSnapshot,
oplog *oplog.OpLog,
peer *v1.Multihost_Peer, // The peer this handler is associated with, must be set before calling OnConnectionEstablished.
syncClient *SyncClient,
) *syncSessionHandlerClient {
return &syncSessionHandlerClient{
l: l,
@@ -186,6 +291,8 @@ func newSyncHandlerClient(
canForwardReposSet: make(map[string]struct{}),
canForwardPlansSet: make(map[string]struct{}),
syncClient: syncClient,
}
}
@@ -368,11 +475,8 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
},
})
}
c.oplog.Subscribe(oplog.Query{}, &oplogSubscription)
go func() {
<-ctx.Done()
c.oplog.Unsubscribe(&oplogSubscription)
}()
c.oplogSubscription = &oplogSubscription
c.oplog.Subscribe(oplog.Query{}, c.oplogSubscription)
// Send initial operation manifest to the server for reconciliation.
opCount, err := c.sendManifest(stream)
@@ -386,6 +490,13 @@ func (c *syncSessionHandlerClient) OnConnectionEstablished(ctx context.Context,
return nil
}
func (c *syncSessionHandlerClient) OnConnectionDisconnected() {
if c.oplogSubscription != nil {
c.oplog.Unsubscribe(c.oplogSubscription)
c.oplogSubscription = nil
}
}
func (c *syncSessionHandlerClient) HandleRequestResources(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionRequestResources) error {
_, _, err := c.sendResourceList(ctx, stream)
return err
@@ -663,3 +774,14 @@ func (c *syncSessionHandlerClient) sendResourceList(ctx context.Context, stream
return len(repoMetadatas), len(planMetadatas), nil
}
func (c *syncSessionHandlerClient) HandleAcquireLockResponse(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionAcquireLockResponse) error {
if c.syncClient != nil {
select {
case c.syncClient.lockResponseCh <- item:
default:
c.l.Warn("lock response channel full, dropping response", zap.String("key", item.GetLockKey()))
}
}
return nil
}
internal/api/syncapi/synccommon.go
+45
View File
@@ -32,6 +32,12 @@ func runSync(
pairingSecret string, // optional one-time pairing secret to send during the handshake
onUnknownPeer onUnknownPeerFunc, // optional callback for handling unknown peers (e.g. pairing), nil to reject all unknown peers
) error {
// Session-scoped context: cancelled when this runSync invocation returns. Any per-session
// goroutines the handler spawns (heartbeats, watchers, etc.) should use this ctx so they
// die with the session rather than outliving it into the next reconnect cycle.
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// send the initial handshake packet to the peer to establish the connection.
handshakePacket, err := createHandshakePacket(localInstanceID, localKey, pairingSecret)
if err != nil {
@@ -118,6 +124,22 @@ func runSync(
if err := handler.HandleReceiveLogData(ctx, commandStream, item.GetReceiveLogData()); err != nil {
return fmt.Errorf("handling receive log data: %w", err)
}
case *v1sync.SyncStreamItem_AcquireLock:
if err := handler.HandleAcquireLock(ctx, commandStream, item.GetAcquireLock()); err != nil {
return fmt.Errorf("handling acquire lock: %w", err)
}
case *v1sync.SyncStreamItem_AcquireLockResponse:
if err := handler.HandleAcquireLockResponse(ctx, commandStream, item.GetAcquireLockResponse()); err != nil {
return fmt.Errorf("handling acquire lock response: %w", err)
}
case *v1sync.SyncStreamItem_ReleaseLock:
if err := handler.HandleReleaseLock(ctx, commandStream, item.GetReleaseLock()); err != nil {
return fmt.Errorf("handling release lock: %w", err)
}
case *v1sync.SyncStreamItem_RefreshLock:
if err := handler.HandleRefreshLock(ctx, commandStream, item.GetRefreshLock()); err != nil {
return fmt.Errorf("handling refresh lock: %w", err)
}
case *v1sync.SyncStreamItem_Throttle:
if err := handler.HandleThrottle(ctx, commandStream, item.GetThrottle()); err != nil {
return fmt.Errorf("handling throttle: %w", err)
@@ -220,6 +242,9 @@ 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.
//
// The ctx passed to every method is scoped to the session: it is cancelled when runSync returns.
// Goroutines spawned by the handler should use this ctx so they don't leak across reconnect cycles.
type syncSessionHandler interface {
OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error
OnConnectionDisconnected()
@@ -234,6 +259,10 @@ type syncSessionHandler interface {
HandleRequestLog(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionRequestLog) error
HandleReceiveLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionReceiveLogData) error
HandleThrottle(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionThrottle) error
HandleAcquireLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionAcquireLock) error
HandleAcquireLockResponse(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionAcquireLockResponse) error
HandleReleaseLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionReleaseLock) error
HandleRefreshLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionRefreshLock) error
}
type unimplementedSyncSessionHandler struct{}
@@ -290,6 +319,22 @@ func (h *unimplementedSyncSessionHandler) HandleThrottle(ctx context.Context, st
return NewSyncErrorProtocol(fmt.Errorf("HandleThrottle not implemented"))
}
func (h *unimplementedSyncSessionHandler) HandleAcquireLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionAcquireLock) error {
return nil // default: ignore lock requests
}
func (h *unimplementedSyncSessionHandler) HandleAcquireLockResponse(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionAcquireLockResponse) error {
return nil // default: ignore lock responses
}
func (h *unimplementedSyncSessionHandler) HandleReleaseLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionReleaseLock) error {
return nil // default: ignore lock releases
}
func (h *unimplementedSyncSessionHandler) HandleRefreshLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionRefreshLock) error {
return nil // default: ignore lock refreshes
}
type remoteOpIdCacheKey struct {
OriginalInstanceKeyid unique.Handle[string]
ID int64
internal/api/syncapi/syncmanager.go
+96 -12
View File
@@ -14,8 +14,10 @@ import (
"github.com/garethgeorge/backrest/internal/config"
"github.com/garethgeorge/backrest/internal/cryptoutil"
"github.com/garethgeorge/backrest/internal/oplog"
hooktypes "github.com/garethgeorge/backrest/internal/hook/types"
"github.com/garethgeorge/backrest/internal/orchestrator"
"go.uber.org/zap"
"google.golang.org/protobuf/proto"
)
// connectedPeerHandle represents a connected peer's stream and metadata.
@@ -54,6 +56,8 @@ type SyncManager struct {
connectedPeers map[string]*connectedPeerHandle
peerStateManager PeerStateManager
lockManager *LockManager
}
func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, orchestrator *orchestrator.Orchestrator, peerStateManager PeerStateManager) *SyncManager {
@@ -81,7 +85,7 @@ func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, orchest
} else {
zap.S().Errorf("syncmanager failed to get initial config: %v", err)
}
return &SyncManager{
mgr := &SyncManager{
configMgr: configMgr,
orchestrator: orchestrator,
oplog: oplog,
@@ -91,7 +95,10 @@ func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, orchest
connectedPeers: make(map[string]*connectedPeerHandle),
peerStateManager: peerStateManager,
lockManager: NewLockManager(),
}
hooktypes.SetSyncLockClientProvider(mgr)
return mgr
}
// GetSyncClients returns a copy of the sync clients map. This makes the map safe to read from concurrently.
@@ -101,6 +108,30 @@ func (m *SyncManager) GetSyncClients() map[string]*SyncClient {
return maps.Clone(m.syncClients)
}
// GetSyncClient returns the sync client for the given instance ID, or nil if not found.
// The map is keyed by Keyid internally (unique and stable), but callers typically only know
// the user-facing InstanceId, so this scans the small set of active clients.
func (m *SyncManager) GetSyncClient(instanceID string) *SyncClient {
m.mu.Lock()
defer m.mu.Unlock()
for _, client := range m.syncClients {
if client.peer.GetInstanceId() == instanceID {
return client
}
}
return nil
}
// GetSyncLockClient returns a SyncLockClient for the given instance ID, satisfying the
// types.SyncLockClientProvider interface for the synclock hook handler.
func (m *SyncManager) GetSyncLockClient(instanceID string) hooktypes.SyncLockClient {
client := m.GetSyncClient(instanceID)
if client == nil {
return nil
}
return client
}
// Note: top level function will be called holding the lock, must kick off goroutines and then return.
func (m *SyncManager) RunSync(ctx context.Context) {
var syncWg sync.WaitGroup
@@ -112,7 +143,7 @@ func (m *SyncManager) RunSync(ctx context.Context) {
zap.L().Info("syncmanager exited")
}()
runSyncWithNewConfig := func() {
startSync := func(config *v1.Config) {
m.mu.Lock()
defer m.mu.Unlock()
@@ -126,12 +157,6 @@ func (m *SyncManager) RunSync(ctx context.Context) {
syncCtx, cancel := context.WithCancel(ctx)
cancelLastSync = cancel
config, err := m.configMgr.Get()
if err != nil {
zap.S().Errorf("syncmanager failed to refresh config with latest changes so sync is stopped: %v", err)
return
}
if config.Multihost.GetIdentity() == nil {
zap.S().Info("syncmanager no identity key configured, sync feature is disabled.")
m.snapshot = nil // Clear the snapshot to indicate sync is disabled
@@ -175,14 +200,68 @@ func (m *SyncManager) RunSync(ctx context.Context) {
}
}
runSyncWithNewConfig()
// lastConfig tracks the config that sync is currently running with.
// We only restart sync when the config changes in a meaningful way
// (i.e. ignoring the Modno field which increments on every write).
var lastConfig *v1.Config
syncConfigEqual := func(a, b *v1.Config) bool {
if a == nil || b == nil {
return a == b
}
// Compare ignoring Modno which changes on every config write.
ac := proto.Clone(a).(*v1.Config)
bc := proto.Clone(b).(*v1.Config)
ac.Modno = 0
bc.Modno = 0
return proto.Equal(ac, bc)
}
restartSyncIfChanged := func() {
config, err := m.configMgr.Get()
if err != nil {
zap.S().Errorf("syncmanager failed to refresh config with latest changes so sync is stopped: %v", err)
return
}
if syncConfigEqual(config, lastConfig) {
zap.L().Debug("syncmanager config changed but sync-relevant config is unchanged, skipping restart")
return
}
lastConfig = proto.Clone(config).(*v1.Config)
startSync(config)
}
restartSyncIfChanged()
// Clock jump detection: if the ticker fires much later than expected
// (e.g. after system sleep), force a reconnect to recover dead streams.
clockJumpInterval := 1 * time.Minute
clockJumpGrace := 30 * time.Second
clockJumpTicker := time.NewTicker(clockJumpInterval)
defer clockJumpTicker.Stop()
lastTickTime := time.Now()
for {
select {
case <-ctx.Done():
return
case <-configWatchCh:
runSyncWithNewConfig()
restartSyncIfChanged()
case <-clockJumpTicker.C:
delta := time.Since(lastTickTime) - clockJumpInterval
lastTickTime = time.Now()
if delta < 0 {
delta = -delta
}
if delta > clockJumpGrace {
zap.S().Warnf("syncmanager detected clock jump of %v, forcing reconnection", delta)
config, err := m.configMgr.Get()
if err != nil {
zap.S().Errorf("syncmanager failed to get config after clock jump: %v", err)
continue
}
startSync(config)
}
}
}
}
@@ -201,13 +280,18 @@ func (m *SyncManager) runSyncWithPeerInternal(ctx context.Context, config *v1.Co
return fmt.Errorf("creating sync client: %w", err)
}
m.mu.Lock()
m.syncClients[knownHostPeer.InstanceId] = newClient
m.syncClients[knownHostPeer.Keyid] = newClient
m.mu.Unlock()
go func() {
newClient.RunSync(ctx)
m.mu.Lock()
delete(m.syncClients, knownHostPeer.InstanceId)
// Only remove the entry if it still points at us. On reconfiguration the new
// client may have already inserted itself under the same key; deleting blindly
// would wipe the replacement.
if m.syncClients[knownHostPeer.Keyid] == newClient {
delete(m.syncClients, knownHostPeer.Keyid)
}
m.mu.Unlock()
}()
internal/api/syncapi/syncserver.go
+54 -1
View File
@@ -29,7 +29,10 @@ type BackrestSyncHandler struct {
var _ v1syncconnect.BackrestSyncServiceHandler = &BackrestSyncHandler{}
func NewBackrestSyncHandler(mgr *SyncManager) *BackrestSyncHandler {
mapper, _ := newRemoteOpIDMapper(mgr.oplog, 4096) // error can be ignored, it just checks for valid size
mapper, err := newRemoteOpIDMapper(mgr.oplog, 4096)
if err != nil {
panic(fmt.Errorf("syncapi: constructing remote op ID mapper: %w", err))
}
return &BackrestSyncHandler{
mgr: mgr,
mapper: mapper,
@@ -230,6 +233,32 @@ func (h *syncSessionHandlerServer) OnConnectionDisconnected() {
}
}
func (h *syncSessionHandlerServer) HandleAcquireLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionAcquireLock) error {
acquired := h.mgr.lockManager.Acquire(item.GetLockKey(), item.GetHolderId())
h.l.Debug("lock acquire request", zap.String("key", item.GetLockKey()), zap.String("holder", item.GetHolderId()), zap.Bool("acquired", acquired))
stream.Send(&v1sync.SyncStreamItem{
Action: &v1sync.SyncStreamItem_AcquireLockResponse{
AcquireLockResponse: &v1sync.SyncStreamItem_SyncActionAcquireLockResponse{
Acquired: acquired,
LockKey: item.GetLockKey(),
},
},
})
return nil
}
func (h *syncSessionHandlerServer) HandleReleaseLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionReleaseLock) error {
released := h.mgr.lockManager.Release(item.GetLockKey(), item.GetHolderId())
h.l.Debug("lock release request", zap.String("key", item.GetLockKey()), zap.String("holder", item.GetHolderId()), zap.Bool("released", released))
return nil
}
func (h *syncSessionHandlerServer) HandleRefreshLock(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionRefreshLock) error {
refreshed := h.mgr.lockManager.Refresh(item.GetLockKey(), item.GetHolderId())
h.l.Debug("lock refresh request", zap.String("key", item.GetLockKey()), zap.String("holder", item.GetHolderId()), zap.Bool("refreshed", refreshed))
return nil
}
func (h *syncSessionHandlerServer) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionHeartbeat) error {
peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone()
if peerState == nil {
@@ -385,6 +414,30 @@ func (h *syncSessionHandlerServer) sendSharedReposToClient(stream *bidiSyncComma
if repo.GetShared() {
repoCopy := proto.Clone(repo).(*v1.Repo)
repoCopy.OriginInstanceId = config.Instance
// Inject lock hooks so the client acquires a lock on this server
// before running operations on the shared repo.
repoCopy.Hooks = append(repoCopy.Hooks,
&v1.Hook{
Conditions: []v1.Hook_Condition{v1.Hook_CONDITION_ANY_START},
OnError: v1.Hook_ON_ERROR_IGNORE,
Action: &v1.Hook_ActionSyncLock{
ActionSyncLock: &v1.Hook_SyncLock{
TargetInstanceId: config.Instance,
LockKey: repo.GetId(),
},
},
},
&v1.Hook{
Conditions: []v1.Hook_Condition{v1.Hook_CONDITION_ANY_END},
OnError: v1.Hook_ON_ERROR_IGNORE,
Action: &v1.Hook_ActionSyncLock{
ActionSyncLock: &v1.Hook_SyncLock{
TargetInstanceId: config.Instance,
LockKey: repo.GetId(),
},
},
},
)
sharedRepos = append(sharedRepos, repoCopy)
}
}
internal/hook/types/synclock.go
+160
View File
@@ -0,0 +1,160 @@
package types
import (
"context"
"fmt"
"reflect"
"sync"
"time"
v1 "github.com/garethgeorge/backrest/gen/go/v1"
"github.com/garethgeorge/backrest/internal/orchestrator/tasks"
"go.uber.org/zap"
)
// SyncLockClient is an interface for acquiring and releasing locks on remote peers.
// This is implemented by syncapi.SyncClient.
type SyncLockClient interface {
AcquireLock(ctx context.Context, lockKey string) (bool, error)
ReleaseLock(lockKey string)
RefreshLock(lockKey string)
}
// SyncLockClientProvider provides SyncLockClients by target instance ID.
type SyncLockClientProvider interface {
GetSyncLockClient(instanceID string) SyncLockClient
}
var (
syncLockProviderMu sync.Mutex
syncLockProvider SyncLockClientProvider
)
// SetSyncLockClientProvider registers the provider used by the synclock hook handler.
func SetSyncLockClientProvider(provider SyncLockClientProvider) {
syncLockProviderMu.Lock()
defer syncLockProviderMu.Unlock()
syncLockProvider = provider
}
func getSyncLockClientProvider() SyncLockClientProvider {
syncLockProviderMu.Lock()
defer syncLockProviderMu.Unlock()
return syncLockProvider
}
const (
lockRefreshInterval = 10 * time.Second
lockMaxRetryDelay = 60 * time.Second
lockInitialRetryDelay = 1 * time.Second
lockMaxRetries = 7 // 1s, 2s, 4s, 8s, 16s, 32s, 60s
)
type syncLockHandler struct{}
func (syncLockHandler) Name() string {
return "synclock"
}
func (syncLockHandler) ActionType() reflect.Type {
return reflect.TypeOf(&v1.Hook_ActionSyncLock{})
}
func (h syncLockHandler) Execute(ctx context.Context, hook *v1.Hook, vars interface{}, runner tasks.TaskRunner, event v1.Hook_Condition) error {
lockConfig := hook.GetActionSyncLock()
if lockConfig == nil {
return fmt.Errorf("synclock hook missing action config")
}
provider := getSyncLockClientProvider()
if provider == nil {
zap.L().Warn("synclock: no provider registered, skipping lock operation")
return nil
}
client := provider.GetSyncLockClient(lockConfig.GetTargetInstanceId())
if client == nil {
zap.L().Warn("synclock: no client for target instance, skipping lock operation",
zap.String("targetInstance", lockConfig.GetTargetInstanceId()))
return nil
}
switch event {
case v1.Hook_CONDITION_ANY_START:
return h.acquireLock(ctx, client, lockConfig)
case v1.Hook_CONDITION_ANY_END:
h.releaseLock(client, lockConfig)
return nil
default:
return nil
}
}
func (h syncLockHandler) acquireLock(ctx context.Context, client SyncLockClient, config *v1.Hook_SyncLock) error {
lockKey := config.GetLockKey()
delay := lockInitialRetryDelay
for attempt := 0; attempt <= lockMaxRetries; attempt++ {
acquired, err := client.AcquireLock(ctx, lockKey)
if err != nil {
zap.L().Warn("synclock: error acquiring lock, proceeding without lock (best-effort)",
zap.String("lockKey", lockKey), zap.Error(err))
return nil // best-effort: proceed without lock
}
if acquired {
zap.L().Info("synclock: acquired lock", zap.String("lockKey", lockKey))
// Start refresh goroutine
go h.refreshLoop(ctx, client, lockKey)
return nil
}
if attempt == lockMaxRetries {
break
}
zap.L().Info("synclock: lock not acquired, retrying",
zap.String("lockKey", lockKey),
zap.Int("attempt", attempt+1),
zap.Duration("delay", delay))
select {
case <-time.After(delay):
case <-ctx.Done():
zap.L().Warn("synclock: context cancelled while waiting for lock, proceeding without lock",
zap.String("lockKey", lockKey))
return nil
}
delay *= 2
if delay > lockMaxRetryDelay {
delay = lockMaxRetryDelay
}
}
zap.L().Warn("synclock: could not acquire lock after retries, proceeding without lock (best-effort)",
zap.String("lockKey", lockKey))
return nil // best-effort: default open
}
func (h syncLockHandler) releaseLock(client SyncLockClient, config *v1.Hook_SyncLock) {
client.ReleaseLock(config.GetLockKey())
zap.L().Info("synclock: released lock", zap.String("lockKey", config.GetLockKey()))
}
func (h syncLockHandler) refreshLoop(ctx context.Context, client SyncLockClient, lockKey string) {
ticker := time.NewTicker(lockRefreshInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
client.RefreshLock(lockKey)
case <-ctx.Done():
return
}
}
}
func init() {
DefaultRegistry().RegisterHandler(&syncLockHandler{})
}
internal/orchestrator/orchestrator.go
+15 -6
View File
@@ -281,12 +281,6 @@ func (o *Orchestrator) ScheduleDefaultTasks(config *v1.Config) error {
return fmt.Errorf("repo %q not found for plan %q", plan.Repo, plan.Id)
}
// Skip plans targeting repos managed by a remote instance; the remote
// instance's orchestrator owns scheduling for those.
if repo.GetOriginInstanceId() != "" {
continue
}
t := tasks.NewScheduledBackupTask(repo, plan)
if err := o.ScheduleTask(t, tasks.TaskPriorityDefault); err != nil {
return fmt.Errorf("schedule backup task for plan %q: %w", plan.Id, err)
@@ -658,8 +652,23 @@ func (o *Orchestrator) cleanupTaskContext(ctx context.Context, op *v1.Operation,
func (o *Orchestrator) executeTask(ctx context.Context, st tasks.ScheduledTask) error {
start := time.Now()
runner := newTaskRunnerImpl(o, st.Task, st.Op)
// Execute ANY_START hooks (best-effort, errors are logged but don't block)
if err := runner.ExecuteHooks(ctx, []v1.Hook_Condition{v1.Hook_CONDITION_ANY_START}, tasks.HookVars{}); err != nil {
runner.Logger(ctx).Warn("ANY_START hook error (best-effort, continuing)", zap.Error(err))
}
err := st.Task.Run(ctx, st, runner)
// Execute ANY_END hooks (best-effort)
endVars := tasks.HookVars{}
if err != nil {
endVars.Error = err.Error()
}
if endErr := runner.ExecuteHooks(ctx, []v1.Hook_Condition{v1.Hook_CONDITION_ANY_END}, endVars); endErr != nil {
runner.Logger(ctx).Warn("ANY_END hook error (best-effort, continuing)", zap.Error(endErr))
}
// Record metrics based on task result
if err != nil {
runner.Logger(ctx).Error("task failed", zap.Error(err), zap.Duration("duration", time.Since(start)))
internal/orchestrator/tasks/hookvars.go
+4
View File
@@ -61,6 +61,10 @@ func (v HookVars) EventName(cond v1.Hook_Condition) string {
return "forget error"
case v1.Hook_CONDITION_FORGET_SUCCESS:
return "forget success"
case v1.Hook_CONDITION_ANY_START:
return "any start"
case v1.Hook_CONDITION_ANY_END:
return "any end"
default:
return "unknown"
}
internal/protoutil/conditions.go
+1
View File
@@ -9,6 +9,7 @@ var startConditionsMap = map[v1.Hook_Condition]bool{
v1.Hook_CONDITION_PRUNE_START: true,
v1.Hook_CONDITION_SNAPSHOT_START: true,
v1.Hook_CONDITION_FORGET_START: true,
v1.Hook_CONDITION_ANY_START: true,
}
var errorConditionsMap = map[v1.Hook_Condition]bool{
proto/v1/config.proto
+10
View File
@@ -202,6 +202,10 @@ message Hook {
CONDITION_FORGET_START = 300; // forget started.
CONDITION_FORGET_ERROR = 301; // forget failed.
CONDITION_FORGET_SUCCESS = 302; // forget succeeded.
// any operation conditions
CONDITION_ANY_START = 400; // before any operation starts.
CONDITION_ANY_END = 401; // after any operation ends (success or fail).
}
enum OnError {
@@ -225,6 +229,7 @@ message Hook {
Shoutrrr action_shoutrrr = 105 [json_name="actionShoutrrr"];
Healthchecks action_healthchecks = 106 [json_name="actionHealthchecks"];
Telegram action_telegram = 107 [json_name="actionTelegram"];
SyncLock action_sync_lock = 109 [json_name="actionSyncLock"];
}
message Command {
@@ -275,6 +280,11 @@ message Hook {
string chat_id = 2 [json_name="chatId"];
string template = 3 [json_name="template"]; // template for the message text.
}
message SyncLock {
string target_instance_id = 1 [json_name="targetInstanceId"]; // the instance ID of the peer to acquire the lock on.
string lock_key = 2 [json_name="lockKey"]; // the lock key, typically the repo ID.
}
}
message Auth {
proto/v1sync/syncservice.proto
+26
View File
@@ -134,6 +134,11 @@ message SyncStreamItem {
SyncActionRequestLog request_log = 30;
SyncActionReceiveLogData receive_log_data = 31;
SyncActionAcquireLock acquire_lock = 40;
SyncActionAcquireLockResponse acquire_lock_response = 41;
SyncActionReleaseLock release_lock = 42;
SyncActionRefreshLock refresh_lock = 43;
SyncActionThrottle throttle = 1000;
}
@@ -217,4 +222,25 @@ message SyncStreamItem {
// See https://pkg.go.dev/crypto/ecdh#PrivateKey.ECDH .
string ecdsa_pub = 2 [json_name="ecdsaPub"]; // base64 encoded public key
}
// Lock protocol messages for coordinating repo access between peers.
message SyncActionAcquireLock {
string lock_key = 1; // the lock key, typically a repo ID.
string holder_id = 2; // the instance ID of the lock requester.
}
message SyncActionAcquireLockResponse {
bool acquired = 1; // whether the lock was acquired.
string lock_key = 2; // the lock key that was requested.
}
message SyncActionReleaseLock {
string lock_key = 1; // the lock key to release.
string holder_id = 2; // the instance ID of the lock holder.
}
message SyncActionRefreshLock {
string lock_key = 1; // the lock key to refresh.
string holder_id = 2; // the instance ID of the lock holder.
}
}
webui/gen/ts/v1/config_pb.ts
+49 -1
View File
File diff suppressed because one or more lines are too long
webui/gen/ts/v1sync/syncservice_pb.ts
+131 -1
View File
File diff suppressed because one or more lines are too long
webui/src/app/App.tsx
+11
View File
@@ -552,6 +552,17 @@ const SidebarRepoItem = React.memo(
>
{repo.id}
</Text>
{repo.originInstanceId && (
<Text
fontSize="xs"
color="fg.muted"
overflow="hidden"
textOverflow="ellipsis"
whiteSpace="nowrap"
>
{repo.originInstanceId}
</Text>
)}
</Box>
</Tooltip>
<Box
webui/src/features/dashboard/SummaryDashboard.tsx
+47 -2
View File
@@ -362,6 +362,7 @@ const MultihostSummary = ({
}: {
multihostConfig: Multihost | null;
}) => {
const [config] = useConfig();
const allPeerStates = useSyncStates();
const peerStates = useMemo(() => {
const map = new Map<string, PeerState>();
@@ -371,6 +372,19 @@ const MultihostSummary = ({
return map;
}, [allPeerStates]);
// Build a map of host instance ID -> repo IDs shared by that host (repos in local config with originInstanceId set)
const sharedReposByHost = useMemo(() => {
const map = new Map<string, string[]>();
for (const repo of config?.repos || []) {
if (repo.originInstanceId) {
const repos = map.get(repo.originInstanceId) || [];
repos.push(repo.id);
map.set(repo.originInstanceId, repos);
}
}
return map;
}, [config?.repos]);
const knownHostTiles: JSX.Element[] = [];
for (const cfgPeer of multihostConfig?.knownHosts || []) {
const peerState = peerStates.get(cfgPeer.keyid);
@@ -378,7 +392,11 @@ const MultihostSummary = ({
continue;
}
knownHostTiles.push(
<PeerStateTile peerState={peerState} key={peerState.peerKeyid} />,
<PeerStateTile
peerState={peerState}
sharedRepoIds={sharedReposByHost.get(peerState.peerInstanceId)}
key={peerState.peerKeyid}
/>,
);
}
@@ -411,7 +429,13 @@ const MultihostSummary = ({
);
};
const PeerStateTile = ({ peerState }: { peerState: PeerState }) => {
const PeerStateTile = ({
peerState,
sharedRepoIds,
}: {
peerState: PeerState;
sharedRepoIds?: string[];
}) => {
const state = useState(1);
useEffect(() => {
// Force rerender every second to update the last heartbeat time
@@ -470,6 +494,27 @@ const PeerStateTile = ({ peerState }: { peerState: PeerState }) => {
}
/>
)}
{sharedRepoIds && sharedRepoIds.length > 0 && (
<DataListItem
label="Shared Repos"
value={
<Flex gap={1} flexWrap="wrap">
{sharedRepoIds.map((repoId) => (
<Box
key={repoId}
px={2}
py={0.5}
bg="bg.muted"
borderRadius="sm"
fontSize="xs"
>
{repoId}
</Box>
))}
</Flex>
}
/>
)}
</DataListRoot>
</Card.Body>
</Card.Root>