Ledger Deltas: AccountDelta retrieval through ledger/catchup service

catchup/service.go

@@ -43,6 +43,9 @@ const blockQueryPeerLimit = 10
 // this should be at least the number of relays
 const catchupRetryLimit = 500
 
+// ErrSyncRoundInvalid is returned when the sync round requested is behind the current ledger round
+var ErrSyncRoundInvalid = errors.New("requested sync round cannot be less than the latest round")
+
 // PendingUnmatchedCertificate is a single certificate that is being waited upon to have its corresponding block fetched.
 type PendingUnmatchedCertificate struct {
 	Cert         agreement.Certificate
@@ -64,7 +67,10 @@ type Ledger interface {
 
 // Service represents the catchup service. Once started and until it is stopped, it ensures that the ledger is up to date with network.
 type Service struct {
-	syncStartNS         int64 // at top of struct to keep 64 bit aligned for atomic.* ops
+	syncStartNS int64 // at top of struct to keep 64 bit aligned for atomic.* ops
+	// disableSyncRound, provided externally, is the first round we will _not_ fetch from the network
+	// any round >= disableSyncRound will not be fetched. If set to 0, it will be disregarded.
+	disableSyncRound    uint64
 	cfg                 config.Local
 	ledger              Ledger
 	ctx                 context.Context
@@ -147,6 +153,26 @@ func (s *Service) IsSynchronizing() (synchronizing bool, initialSync bool) {
 	return
 }
 
+// SetDisableSyncRound attempts to set the first round we _do_not_ want to fetch from the network
+// Blocks from disableSyncRound or any round after disableSyncRound will not be fetched while this is set
+func (s *Service) SetDisableSyncRound(rnd uint64) error {
+	if basics.Round(rnd) < s.ledger.LastRound() {
+		return ErrSyncRoundInvalid
+	}
+	atomic.StoreUint64(&s.disableSyncRound, rnd)
+	return nil
+}
+
+// UnsetDisableSyncRound removes any previously set disabled sync round
+func (s *Service) UnsetDisableSyncRound() {
+	atomic.StoreUint64(&s.disableSyncRound, 0)
+}
+
+// GetDisableSyncRound returns the disabled sync round
+func (s *Service) GetDisableSyncRound() uint64 {
+	return atomic.LoadUint64(&s.disableSyncRound)
+}
+
 // SynchronizingTime returns the time we've been performing a catchup operation (0 if not currently catching up)
 func (s *Service) SynchronizingTime() time.Duration {
 	startNS := atomic.LoadInt64(&s.syncStartNS)
@@ -202,6 +228,10 @@ func (s *Service) innerFetch(r basics.Round, peer network.Peer) (blk *bookkeepin
 //  - If the block is already in the ledger (e.g. if agreement service has already written it)
 //  - If the retrieval of the previous block was unsuccessful
 func (s *Service) fetchAndWrite(r basics.Round, prevFetchCompleteChan chan bool, lookbackComplete chan bool, peerSelector *peerSelector) bool {
+	// If sync-ing this round is not intended, don't fetch it
+	if dontSyncRound := s.GetDisableSyncRound(); dontSyncRound != 0 && r >= basics.Round(dontSyncRound) {
+		return false
+	}
 	i := 0
 	hasLookback := false
 	for true {

catchup/service_test.go

@@ -181,6 +181,96 @@ func (cl *periodicSyncLogger) Warnf(s string, args ...interface{}) {
 	cl.Logger.Warnf(s, args...)
 }
 
+func TestSyncRound(t *testing.T) {
+	partitiontest.PartitionTest(t)
+
+	// Make Ledger
+	local := new(mockedLedger)
+	local.blocks = append(local.blocks, bookkeeping.Block{})
+
+	remote, _, blk, err := buildTestLedger(t, bookkeeping.Block{})
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+	addBlocks(t, remote, blk, 10)
+
+	// Create a network and block service
+	blockServiceConfig := config.GetDefaultLocal()
+	net := &httpTestPeerSource{}
+	ls := rpcs.MakeBlockService(logging.Base(), blockServiceConfig, remote, net, "test genesisID")
+
+	nodeA := basicRPCNode{}
+	nodeA.RegisterHTTPHandler(rpcs.BlockServiceBlockPath, ls)
+	nodeA.start()
+	defer nodeA.stop()
+	rootURL := nodeA.rootURL()
+	net.addPeer(rootURL)
+
+	auth := &mockedAuthenticator{fail: true}
+	initialLocalRound := local.LastRound()
+	require.True(t, 0 == initialLocalRound)
+
+	// Make Service
+	localCfg := config.GetDefaultLocal()
+	s := MakeService(logging.Base(), localCfg, net, local, auth, nil, nil)
+	s.log = &periodicSyncLogger{Logger: logging.Base()}
+	s.deadlineTimeout = 2 * time.Second
+
+	// Set disable round success
+	err = s.SetDisableSyncRound(3)
+	require.NoError(t, err)
+
+	s.Start()
+	defer s.Stop()
+	// wait past the initial sync - which is known to fail due to the above "auth"
+	time.Sleep(s.deadlineTimeout*2 - 200*time.Millisecond)
+	require.Equal(t, initialLocalRound, local.LastRound())
+	auth.alter(-1, false)
+
+	// wait until the catchup is done. Since we've might have missed the sleep window, we need to wait
+	// until the synchronization is complete.
+	waitStart := time.Now()
+	for time.Since(waitStart) < 2*s.deadlineTimeout {
+		if remote.LastRound() == local.LastRound() {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	// Assert that the last block is the one we expect--i.e. disableSyncRound - 1
+	rnd := s.GetDisableSyncRound()
+	rr, lr := basics.Round(rnd-1), local.LastRound()
+	require.Equal(t, rr, lr)
+
+	for r := basics.Round(1); r < rr; r++ {
+		localBlock, err := local.Block(r)
+		require.NoError(t, err)
+		remoteBlock, err := remote.Block(r)
+		require.NoError(t, err)
+		require.Equal(t, remoteBlock.Hash(), localBlock.Hash())
+	}
+
+	// unset syncRound and make sure we finish catching up
+	s.UnsetDisableSyncRound()
+	// wait until the catchup is done
+	waitStart = time.Now()
+	for time.Now().Sub(waitStart) < 8*s.deadlineTimeout {
+		if remote.LastRound() == local.LastRound() {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	rr, lr = remote.LastRound(), local.LastRound()
+	require.Equal(t, rr, lr)
+	for r := basics.Round(1); r < remote.LastRound(); r++ {
+		localBlock, err := local.Block(r)
+		require.NoError(t, err)
+		remoteBlock, err := remote.Block(r)
+		require.NoError(t, err)
+		require.Equal(t, remoteBlock.Hash(), localBlock.Hash())
+	}
+}
+
 func TestPeriodicSync(t *testing.T) {
 	partitiontest.PartitionTest(t)
 

daemon/algod/api/Makefile

@@ -2,7 +2,7 @@ GOPATH	:= $(shell go env GOPATH)
 GOPATH1	:= $(firstword $(subst :, ,$(GOPATH)))
 
 # `make all` or just `make` should be appropriate for dev work
-all:	server/v2/generated/model/types.go server/v2/generated/nonparticipating/public/routes.go server/v2/generated/nonparticipating/private/routes.go server/v2/generated/participating/public/routes.go server/v2/generated/participating/private/routes.go
+all:	server/v2/generated/model/types.go server/v2/generated/nonparticipating/public/routes.go server/v2/generated/nonparticipating/private/routes.go server/v2/generated/participating/public/routes.go server/v2/generated/participating/private/routes.go server/v2/generated/data/routes.go
 
 # `make generate` should be able to replace old `generate.sh` script and be appropriate for build system use
 generate:	oapi-codegen all
@@ -19,6 +19,10 @@ server/v2/generated/participating/public/routes.go:	algod.oas3.yml
 server/v2/generated/participating/private/routes.go:	algod.oas3.yml
 	$(GOPATH1)/bin/oapi-codegen -config ./server/v2/generated/participating/private/private_routes.yml algod.oas3.yml
 
+
+server/v2/generated/data/routes.go:	algod.oas3.yml
+	$(GOPATH1)/bin/oapi-codegen -config ./server/v2/generated/data/data_routes.yml algod.oas3.yml
+
 server/v2/generated/model/types.go:	algod.oas3.yml
 	$(GOPATH1)/bin/oapi-codegen -config ./server/v2/generated/model/model_types.yml algod.oas3.yml
 
@@ -31,6 +35,6 @@ oapi-codegen:	.PHONY
 	../../../scripts/buildtools/install_buildtools.sh -o github.com/algorand/oapi-codegen -c github.com/algorand/oapi-codegen/cmd/oapi-codegen
 
 clean:
-	rm -rf server/v2/generated/model/types.go server/v2/generated/nonparticipating/public/routes.go server/v2/generated/nonparticipating/private/routes.go server/v2/generated/participating/public/routes.go server/v2/generated/participating/private/routes.go algod.oas3.yml
+	rm -rf server/v2/generated/model/types.go server/v2/generated/nonparticipating/public/routes.go server/v2/generated/nonparticipating/private/routes.go server/v2/generated/participating/public/routes.go server/v2/generated/participating/private/routes.go server/v2/generated/data/routes.go algod.oas3.yml
 
 .PHONY: