introduce entity that will keep info about services current state and integrate it into DASer and Syncer

das/daser.go

@@ -9,13 +9,16 @@ import (
 	logging "github.com/ipfs/go-log/v2"
 
 	"github.com/celestiaorg/celestia-node/header"
+	"github.com/celestiaorg/celestia-node/libs/service"
 	"github.com/celestiaorg/celestia-node/service/share"
 )
 
 var log = logging.Logger("das")
 
 // DASer continuously validates availability of data committed to headers.
 type DASer struct {
+	service.Service
+
 	da   share.Availability
 	hsub header.Subscriber
 
@@ -73,6 +76,7 @@ func (d *DASer) Start(context.Context) error {
 	dasCtx, cancel := context.WithCancel(context.Background())
 	d.cancel = cancel
 
+	d.SetState(service.Running)
 	// kick off catch-up routine manager
 	go d.catchUpManager(dasCtx, checkpoint)
 	// kick off sampling routine for recently received headers
@@ -82,6 +86,11 @@ func (d *DASer) Start(context.Context) error {
 
 // Stop stops sampling.
 func (d *DASer) Stop(ctx context.Context) error {
+	if d.State() == service.Stopped {
+		log.Debug("Daser is stopped")
+		return nil
+	}
+	defer d.SetState(service.Stopped)
 	d.cancel()
 	// wait for both sampling routines to exit
 	for i := 0; i < 2; i++ {

header/sync/sync.go

@@ -13,6 +13,7 @@ import (
 	tmbytes "github.com/tendermint/tendermint/libs/bytes"
 
 	"github.com/celestiaorg/celestia-node/header"
+	"github.com/celestiaorg/celestia-node/libs/service"
 )
 
 var log = logging.Logger("header/sync")
@@ -32,6 +33,8 @@ var log = logging.Logger("header/sync")
 //    	* adds the header to pending cache(making it the latest known trusted header)
 //      * and triggers syncing loop to catch up to that point.
 type Syncer struct {
+	service.Service
+
 	sub      header.Subscriber
 	exchange header.Exchange
 	store    header.Store
@@ -69,6 +72,7 @@ func (s *Syncer) Start(context.Context) error {
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
+	s.SetState(service.Running)
 	go s.syncLoop(ctx)
 	s.wantSync()
 	s.cancel = cancel
@@ -77,6 +81,11 @@ func (s *Syncer) Start(context.Context) error {
 
 // Stop stops Syncer.
 func (s *Syncer) Stop(context.Context) error {
+	if s.Service.State() == service.Stopped {
+		log.Debug("Syncer is stopped")
+		return nil
+	}
+	s.SetState(service.Stopped)
 	s.cancel()
 	s.cancel = nil
 	return nil

libs/service/service.go

@@ -0,0 +1,25 @@
+package service
+
+import "sync/atomic"
+
+type State uint32
+
+const (
+	Stopped State = iota
+	Running
+)
+
+// Service provides an interface to change an check service state
+type Service struct {
+	state uint32
+}
+
+// SetState allows to change service state
+func (s *Service) SetState(state State) {
+	atomic.StoreUint32(&s.state, uint32(state))
+}
+
+// State returns service current state
+func (s *Service) State() State {
+	return State(atomic.LoadUint32(&s.state))
+}