From 957a43a12408fd9e4751d8b72d29bd08f4103ac0 Mon Sep 17 00:00:00 2001
From: Dmitrii Stepanov <d.stepanov@yadro.com>
Date: Tue, 13 Jun 2023 11:43:25 +0300
Subject: [PATCH] [#266] services/tree: Add sync check

Do not accept requests until initial sync is finished.
`Apply` is deliberately left out -- we don't want to miss anything new.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
---
 pkg/services/tree/service.go | 39 ++++++++++++++++++++++++++++++++++++
 pkg/services/tree/sync.go    |  3 ++-
 2 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/pkg/services/tree/service.go b/pkg/services/tree/service.go
index 4364095e2..12d970c42 100644
--- a/pkg/services/tree/service.go
+++ b/pkg/services/tree/service.go
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
@@ -31,6 +32,8 @@ type Service struct {
 	syncChan chan struct{}
 	syncPool *ants.Pool
 
+	initialSyncDone atomic.Bool
+
 	// cnrMap contains existing (used) container IDs.
 	cnrMap map[cidSDK.ID]struct{}
 	// cnrMapMtx protects cnrMap
@@ -90,6 +93,10 @@ func (s *Service) Shutdown() {
 }
 
 func (s *Service) Add(ctx context.Context, req *AddRequest) (*AddResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -138,6 +145,10 @@ func (s *Service) Add(ctx context.Context, req *AddRequest) (*AddResponse, error
 }
 
 func (s *Service) AddByPath(ctx context.Context, req *AddByPathRequest) (*AddByPathResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -198,6 +209,10 @@ func (s *Service) AddByPath(ctx context.Context, req *AddByPathRequest) (*AddByP
 }
 
 func (s *Service) Remove(ctx context.Context, req *RemoveRequest) (*RemoveResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -247,6 +262,10 @@ func (s *Service) Remove(ctx context.Context, req *RemoveRequest) (*RemoveRespon
 // Move applies client operation to the specified tree and pushes in queue
 // for replication on other nodes.
 func (s *Service) Move(ctx context.Context, req *MoveRequest) (*MoveResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -295,6 +314,10 @@ func (s *Service) Move(ctx context.Context, req *MoveRequest) (*MoveResponse, er
 }
 
 func (s *Service) GetNodeByPath(ctx context.Context, req *GetNodeByPathRequest) (*GetNodeByPathResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -371,6 +394,10 @@ func (s *Service) GetNodeByPath(ctx context.Context, req *GetNodeByPathRequest)
 }
 
 func (s *Service) GetSubTree(req *GetSubTreeRequest, srv TreeService_GetSubTreeServer) error {
+	if !s.initialSyncDone.Load() {
+		return ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -500,6 +527,10 @@ func (s *Service) Apply(_ context.Context, req *ApplyRequest) (*ApplyResponse, e
 }
 
 func (s *Service) GetOpLog(req *GetOpLogRequest, srv TreeService_GetOpLogServer) error {
+	if !s.initialSyncDone.Load() {
+		return ErrAlreadySyncing
+	}
+
 	b := req.GetBody()
 
 	var cid cidSDK.ID
@@ -560,6 +591,10 @@ func (s *Service) GetOpLog(req *GetOpLogRequest, srv TreeService_GetOpLogServer)
 }
 
 func (s *Service) TreeList(ctx context.Context, req *TreeListRequest) (*TreeListResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	var cid cidSDK.ID
 
 	err := cid.Decode(req.GetBody().GetContainerId())
@@ -643,5 +678,9 @@ func (s *Service) getContainerInfo(cid cidSDK.ID, pub []byte) ([]netmapSDK.NodeI
 }
 
 func (s *Service) Healthcheck(context.Context, *HealthcheckRequest) (*HealthcheckResponse, error) {
+	if !s.initialSyncDone.Load() {
+		return nil, ErrAlreadySyncing
+	}
+
 	return new(HealthcheckResponse), nil
 }
diff --git a/pkg/services/tree/sync.go b/pkg/services/tree/sync.go
index d132faf6e..e44e8dbbf 100644
--- a/pkg/services/tree/sync.go
+++ b/pkg/services/tree/sync.go
@@ -388,7 +388,7 @@ func (s *Service) syncLoop(ctx context.Context) {
 				s.log.Error(logs.TreeCouldNotFetchContainers, zap.Error(err))
 				s.metrics.AddSyncDuration(time.Since(start), false)
 				span.End()
-				continue
+				break
 			}
 
 			newMap, cnrsToSync := s.containersToSync(cnrs)
@@ -402,6 +402,7 @@ func (s *Service) syncLoop(ctx context.Context) {
 			s.metrics.AddSyncDuration(time.Since(start), true)
 			span.End()
 		}
+		s.initialSyncDone.Store(true)
 	}
 }