Repository: celestiaorg/celestia-node
Pull request #1815 (2023-02-23 18:52:41)

feat(das): Don't sample recent jobs twice
## Overview

 - Prevent DASer from doing same job twice in case of recent sampling is run on synced node.
 - Log recent jobs with info level



Patch:
diff --git a/das/checkpoint.go b/das/checkpoint.go
index 04e6cafa..860c72cb 100644
--- a/das/checkpoint.go
+++ b/das/checkpoint.go
@@ -48,12 +48,3 @@ func (c checkpoint) String() string {
 
 	return str
 }
-
-// totalSampled returns the total amount of sampled headers
-func (c checkpoint) totalSampled() uint64 {
-	var totalInProgress uint64
-	for _, w := range c.Workers {
-		totalInProgress += (w.To - w.From) + 1
-	}
-	return c.SampleFrom - totalInProgress - uint64(len(c.Failed))
-}
diff --git a/das/coordinator.go b/das/coordinator.go
index bb2d82d6..2aca5483 100644
--- a/das/coordinator.go
+++ b/das/coordinator.go
@@ -63,9 +63,6 @@ func newSamplingCoordinator(
 func (sc *samplingCoordinator) run(ctx context.Context, cp checkpoint) {
 	sc.state.resumeFromCheckpoint(cp)
 
-	// the amount of sampled headers from the last checkpoint
-	sc.metrics.recordTotalSampled(cp.totalSampled())
-
 	// resume workers
 	for _, wk := range cp.Workers {
 		sc.runWorker(ctx, sc.state.newJob(wk.From, wk.To))
diff --git a/das/metrics.go b/das/metrics.go
index b11dcc8e..2a4be8fe 100644
--- a/das/metrics.go
+++ b/das/metrics.go
@@ -25,8 +25,7 @@ type metrics struct {
 	getHeaderTime syncfloat64.Histogram
 	newHead       syncint64.Counter
 
-	lastSampledTS   uint64
-	totalSampledInt uint64
+	lastSampledTS uint64
 }
 
 func (d *DASer) InitMetrics() error {
@@ -117,8 +116,7 @@ func (d *DASer) InitMetrics() error {
 				lastSampledTS.Observe(ctx, int64(ts))
 			}
 
-			totalSampledInt := atomic.LoadUint64(&d.sampler.metrics.totalSampledInt)
-			totalSampled.Observe(ctx, int64(totalSampledInt))
+			totalSampled.Observe(ctx, int64(stats.totalSampled()))
 		},
 	)
 
@@ -152,12 +150,6 @@ func (m *metrics) observeSample(
 	)
 
 	atomic.StoreUint64(&m.lastSampledTS, uint64(time.Now().UTC().Unix()))
-
-	// only increment the counter if it's not a recent header job
-	// as those happen twice.
-	if err == nil && !isRecentHeader {
-		atomic.AddUint64(&m.totalSampledInt, 1)
-	}
 }
 
 // observeGetHeader records the time it took to get a header from the header store.
@@ -175,11 +167,3 @@ func (m *metrics) observeNewHead(ctx context.Context) {
 	}
 	m.newHead.Add(ctx, 1)
 }
-
-// recordTotalSampled records the total sampled headers.
-func (m *metrics) recordTotalSampled(totalSampled uint64) {
-	if m == nil {
-		return
-	}
-	atomic.StoreUint64(&m.totalSampledInt, totalSampled)
-}
diff --git a/das/state.go b/das/state.go
index e45a8ab0..8e854e7f 100644
--- a/das/state.go
+++ b/das/state.go
@@ -94,13 +94,18 @@ func (s *coordinatorState) updateHead(newHead int64) {
 }
 
 func (s *coordinatorState) newRecentJob(header *header.ExtendedHeader) job {
+	height := uint64(header.Height())
+	// move next, to prevent catchup job from processing same height
+	if s.next == height {
+		s.next++
+	}
 	s.nextJobID++
 	return job{
 		id:             s.nextJobID,
 		isRecentHeader: true,
 		header:         header,
-		From:           uint64(header.Height()),
-		To:             uint64(header.Height()),
+		From:           height,
+		To:             height,
 	}
 }
 
diff --git a/das/stats.go b/das/stats.go
index 3b0a61ea..a07e8fad 100644
--- a/das/stats.go
+++ b/das/stats.go
@@ -1,21 +1,19 @@
 package das
 
-// SamplingStats collects information about the DASer process. Currently, there are
-// only two sampling routines: the main sampling routine which performs sampling
-// over current network headers, and the `catchUp` routine which performs sampling
-// over past headers from the last sampled checkpoint.
+// SamplingStats collects information about the DASer process.
 type SamplingStats struct {
 	// all headers before SampledChainHead were successfully sampled
 	SampledChainHead uint64 `json:"head_of_sampled_chain"`
-	// all headers before CatchupHead were submitted to sampling workers
+	// all headers before CatchupHead were submitted to sampling workers. They could be either already
+	// sampled, failed or still in progress. For in progress items check Workers stat.
 	CatchupHead uint64 `json:"head_of_catchup"`
 	// NetworkHead is the height of the most recent header in the network
 	NetworkHead uint64 `json:"network_head_height"`
-	// Failed contains all skipped header's heights with corresponding try count
+	// Failed contains all skipped headers heights with corresponding try count
 	Failed map[uint64]int `json:"failed,omitempty"`
 	// Workers has information about each currently running worker stats
 	Workers []WorkerStats `json:"workers,omitempty"`
-	// Concurrency currently running parallel workers
+	// Concurrency amount of currently running parallel workers
 	Concurrency int `json:"concurrency"`
 	// CatchUpDone indicates whether all known headers are sampled
 	CatchUpDone bool `json:"catch_up_done"`
@@ -30,3 +28,12 @@ type WorkerStats struct {
 
 	ErrMsg string `json:"error,omitempty"`
 }
+
+// totalSampled returns the total amount of sampled headers
+func (s SamplingStats) totalSampled() uint64 {
+	var inProgress uint64
+	for _, w := range s.Workers {
+		inProgress += w.To - w.Curr + 1
+	}
+	return s.CatchupHead - inProgress - uint64(len(s.Failed))
+}
diff --git a/das/worker.go b/das/worker.go
index eb416699..516cd833 100644
--- a/das/worker.go
+++ b/das/worker.go
@@ -80,16 +80,13 @@ func (w *worker) run(ctx context.Context, timeout time.Duration, resultCh chan<-
 		}
 	}
 
-	if w.state.Curr > w.state.From {
-		jobTime := time.Since(jobStart)
-		log.Infow(
-			"sampled headers",
-			"from", w.state.From,
-			"to", w.state.Curr,
-			"finished (s)",
-			jobTime.Seconds(),
-		)
-	}
+	log.Infow(
+		"finished sampling headers",
+		"from", w.state.From,
+		"to", w.state.Curr,
+		"errors", len(w.state.failed),
+		"finished (s)", time.Since(jobStart),
+	)
 
 	select {
 	case resultCh <- result{


Test patch:
diff --git a/das/daser_test.go b/das/daser_test.go
index cba84a06..8d0a33e0 100644
--- a/das/daser_test.go
+++ b/das/daser_test.go
@@ -204,12 +204,9 @@ func TestDASerSampleTimeout(t *testing.T) {
 	f := new(fraud.DummyService)
 
 	// create and start DASer
-	daser, err := NewDASer(avail, sub, getter, ds, f, newBroadcastMock(1))
+	daser, err := NewDASer(avail, sub, getter, ds, f, newBroadcastMock(1), WithSampleTimeout(1))
 	require.NoError(t, err)
 
-	// assign directly to avoid params validation
-	daser.params.SampleTimeout = 0
-
 	require.NoError(t, daser.Start(ctx))
 	require.NoError(t, daser.sampler.state.waitCatchUp(ctx))