Repository: celestiaorg/celestia-node
Pull request #910 (2022-07-15 12:11:20)

cmd | node: Refactor core flags
## Description
This PR consolidates flags related to adding a core endpoint. The reasoning behind this is to prevent users from having to pass too many flags once #571 is resolved (since the core accessor will now need both a `gRPC` and `RPC` endpoint for the celestia-core connection).

## New flags

**In the case that the default ports are exposed (grpc:9090, rpc:26657) on the celestia-core endpoint**:
```
celestia <node_type> start --core <ip addr of core endpoint>
```
**In the case that custom ports are exposed on the celestia-core endpoint**:
```
celestia <node_type> start --core <ip addr of core endpoint> --core.grpc 9090 --core.rpc 26657
```

If `--core.grpc` || `--core.rpc` are not provided, we assume the defaults and attempt to connect.




Patch:
diff --git a/README.md b/README.md
index eacf682d..b6a4234f 100644
--- a/README.md
+++ b/README.md
@@ -2,8 +2,7 @@
 
 [![Go Reference](https://pkg.go.dev/badge/github.com/celestiaorg/celestia-node.svg)](https://pkg.go.dev/github.com/celestiaorg/celestia-node)
 [![GitHub release (latest by date including pre-releases)](https://img.shields.io/github/v/release/celestiaorg/celestia-node)](https://github.com/celestiaorg/celestia-node/releases/latest)
-[![test](https://github.com/celestiaorg/celestia-node/actions/workflows/test.yml/badge.svg)](https://github.com/celestiaorg/celestia-node/actions/workflows/test.yml)
-[![lint](https://github.com/celestiaorg/celestia-node/actions/workflows/lint.yml/badge.svg)](https://github.com/celestiaorg/celestia-node/actions/workflows/lint.yml)
+[![Go CI](https://github.com/celestiaorg/celestia-node/actions/workflows/go-ci.yml/badge.svg)](https://github.com/celestiaorg/celestia-node/actions/workflows/go-ci.yml)
 [![Go Report Card](https://goreportcard.com/badge/github.com/celestiaorg/celestia-node)](https://goreportcard.com/report/github.com/celestiaorg/celestia-node)
 [![codecov](https://codecov.io/gh/celestiaorg/celestia-node/branch/main/graph/badge.svg?token=CWGA4RLDS9)](https://codecov.io/gh/celestiaorg/celestia-node)
 
diff --git a/cmd/flags_core.go b/cmd/flags_core.go
index c68ef36a..a42c9fe6 100644
--- a/cmd/flags_core.go
+++ b/cmd/flags_core.go
@@ -2,8 +2,8 @@ package cmd
 
 import (
 	"fmt"
-	"net"
-	"net/url"
+	"strconv"
+	"strings"
 
 	"github.com/spf13/cobra"
 	flag "github.com/spf13/pflag"
@@ -12,8 +12,9 @@ import (
 )
 
 var (
-	coreRemoteFlag = "core.remote"
-	coreGRPCFlag   = "core.grpc"
+	coreFlag     = "core.ip"
+	coreRPCFlag  = "core.rpc.port"
+	coreGRPCFlag = "core.grpc.port"
 )
 
 // CoreFlags gives a set of hardcoded Core flags.
@@ -21,16 +22,20 @@ func CoreFlags() *flag.FlagSet {
 	flags := &flag.FlagSet{}
 
 	flags.String(
-		coreRemoteFlag,
+		coreFlag,
 		"",
-		"Indicates node to connect to the given remote core node. "+
-			"Example: <protocol>://<ip>:<port>, tcp://127.0.0.1:26657",
+		"Indicates node to connect to the given core node. "+
+			"Example: <ip>, 127.0.0.1. Assumes RPC port 26657 and gRPC port 9009 as default unless otherwise specified.",
+	)
+	flags.String(
+		coreRPCFlag,
+		"26657",
+		"Set a custom RPC port for the core node connection. The --core.ip flag must also be provided.",
 	)
 	flags.String(
 		coreGRPCFlag,
-		"",
-		"Indicates node to connect to the given celestia-app gRPC endpoint for state-related queries"+
-			"Example: <ip>:<port>, 127.0.0.1:9090",
+		"9090",
+		"Set a custom gRPC port for the core node connection. The --core.ip flag must also be provided.",
 	)
 
 	return flags
@@ -38,44 +43,47 @@ func CoreFlags() *flag.FlagSet {
 
 // ParseCoreFlags parses Core flags from the given cmd and applies values to Env.
 func ParseCoreFlags(cmd *cobra.Command, env *Env) error {
-	coreRemote := cmd.Flag(coreRemoteFlag).Value.String()
-	if coreRemote != "" {
-		proto, addr, err := validateAddress(coreRemote)
-		if err != nil {
-			return fmt.Errorf("cmd: while parsing '%s': %w", coreRemoteFlag, err)
+	coreIP := cmd.Flag(coreFlag).Value.String()
+	if coreIP == "" {
+		if cmd.Flag(coreGRPCFlag).Changed || cmd.Flag(coreRPCFlag).Changed {
+			return fmt.Errorf("cannot specify RPC/gRPC ports without specifying an IP address for --core.ip")
 		}
-
-		env.AddOptions(node.WithRemoteCore(proto, addr))
+		return nil
 	}
-
-	grpcEndpoint := cmd.Flag(coreGRPCFlag).Value.String()
-	if grpcEndpoint != "" {
-		// parse ip:port from the endpoint
-		grpcURL, err := url.Parse(grpcEndpoint)
-		if err != nil {
-			return err
-		}
-		env.AddOptions(node.WithGRPCEndpoint(grpcURL.Host))
+	// sanity check given core ip addr and strip leading protocol
+	ip, err := sanityCheckIP(coreIP)
+	if err != nil {
+		return err
 	}
 
-	return nil
-}
-
-// validateAddress parses the given address of the remote core node
-// and checks if it configures correctly
-func validateAddress(address string) (string, string, error) {
-	u, err := url.Parse(address)
+	rpc := cmd.Flag(coreRPCFlag).Value.String()
+	// sanity check rpc endpoint
+	_, err = strconv.Atoi(rpc)
 	if err != nil {
-		return "", "", err
+		return err
 	}
+	env.AddOptions(node.WithRemoteCoreIP(ip), node.WithRemoteCorePort(rpc))
 
-	if u.Scheme == "" || u.Host == "" {
-		return "", "", fmt.Errorf("both protocol and host must present in the address")
+	grpc := cmd.Flag(coreGRPCFlag).Value.String()
+	// sanity check gRPC endpoint
+	_, err = strconv.Atoi(grpc)
+	if err != nil {
+		return err
 	}
+	env.AddOptions(node.WithGRPCPort(grpc))
+	return nil
+}
 
-	if _, _, err := net.SplitHostPort(u.Host); err != nil {
-		return "", "", fmt.Errorf("incorrect address provided for Remote Core")
+// sanityCheckIP trims leading protocol scheme and port from the given
+// IP address if present.
+func sanityCheckIP(ip string) (string, error) {
+	original := ip
+	ip = strings.TrimPrefix(ip, "http://")
+	ip = strings.TrimPrefix(ip, "https://")
+	ip = strings.TrimPrefix(ip, "tcp://")
+	ip = strings.Split(ip, ":")[0]
+	if ip == "" {
+		return "", fmt.Errorf("invalid IP addr given: %s", original)
 	}
-
-	return u.Scheme, u.Host, nil
+	return ip, nil
 }
diff --git a/core/client.go b/core/client.go
index 5ad7a102..0f149571 100644
--- a/core/client.go
+++ b/core/client.go
@@ -13,14 +13,14 @@ import (
 type Client = client.Client
 
 // NewRemote creates a new Client that communicates with a remote Core endpoint over HTTP.
-func NewRemote(protocol, remoteAddr string) (Client, error) {
+func NewRemote(ip, port string) (Client, error) {
 	httpClient := retryhttp.NewClient()
 	httpClient.RetryMax = 2
 	// suppress logging
 	httpClient.Logger = nil
 
 	return http.NewWithClient(
-		fmt.Sprintf("%s://%s", protocol, remoteAddr),
+		fmt.Sprintf("tcp://%s:%s", ip, port),
 		httpClient.StandardClient(),
 	)
 }
diff --git a/das/daser.go b/das/daser.go
index 696de672..3676c289 100644
--- a/das/daser.go
+++ b/das/daser.go
@@ -19,19 +19,17 @@ var log = logging.Logger("das")
 
 // DASer continuously validates availability of data committed to headers.
 type DASer struct {
-	da    share.Availability
-	bcast fraud.Broadcaster
-	hsub  header.Subscriber
-
-	// getter allows the DASer to fetch an ExtendedHeader (EH) at a certain height
-	// and blocks until the EH has been processed by the header store.
-	getter header.Getter
-	// checkpoint store
-	cstore datastore.Datastore
-
 	ctx    context.Context
 	cancel context.CancelFunc
 
+	da     share.Availability
+	bcast  fraud.Broadcaster
+	hsub   header.Subscriber // listens for new headers in the network
+	getter header.Getter     // retrieves past headers
+
+	cstore datastore.Datastore // checkpoint store
+	state  state
+
 	jobsCh chan *catchUpJob
 
 	sampleDn  chan struct{} // done signal for sample loop
@@ -113,8 +111,12 @@ func (d *DASer) Stop(ctx context.Context) error {
 
 // sample validates availability for each Header received from header subscription.
 func (d *DASer) sample(ctx context.Context, sub header.Subscription, checkpoint int64) {
+	// indicate sampling routine is running
+	d.indicateRunning()
 	defer func() {
 		sub.Cancel()
+		// indicate sampling routine is stopped
+		d.indicateStopped()
 		// send done signal
 		d.sampleDn <- struct{}{}
 	}()
@@ -153,9 +155,14 @@ func (d *DASer) sample(ctx context.Context, sub header.Subscription, checkpoint
 
 		err = d.sampleHeader(ctx, h)
 		if err != nil {
+			// record error
+			d.updateSampleState(h, err)
+			log.Warn("DASer SAMPLING ROUTINE WILL BE STOPPED. IN ORDER TO CONTINUE SAMPLING, " +
+				"RE-START THE NODE")
 			return
 		}
 
+		d.updateSampleState(h, nil)
 		sampleHeight = h.Height
 	}
 }
@@ -187,13 +194,22 @@ func (d *DASer) catchUpManager(ctx context.Context, checkpoint int64) {
 		case <-ctx.Done():
 			return
 		case job := <-d.jobsCh:
+			// record details of incoming job
+			d.recordJobDetails(job)
 			// perform catchUp routine
 			height, err := d.catchUp(ctx, job)
 			// store the height of the last successfully sampled header
 			checkpoint = height
 			// exit routine if a catch-up job was unsuccessful
 			if err != nil {
+				// record error
+				d.state.catchUpLk.Lock()
+				d.state.catchUp.Error = err
+				d.state.catchUpLk.Unlock()
+
 				log.Errorw("catch-up routine failed", "attempted range: from", job.from, "to", job.to)
+				log.Warn("DASer CATCH-UP SAMPLING ROUTINE WILL BE STOPPED. IN ORDER TO CONTINUE SAMPLING, " +
+					"RE-START THE NODE")
 				return
 			}
 		}
@@ -203,7 +219,6 @@ func (d *DASer) catchUpManager(ctx context.Context, checkpoint int64) {
 // catchUp starts a sampling routine for headers starting at the next header
 // after the `from` height and exits the loop once `to` is reached. (from:to]
 func (d *DASer) catchUp(ctx context.Context, job *catchUpJob) (int64, error) {
-	routineStartTime := time.Now()
 	log.Infow("sampling past headers", "from", job.from, "to", job.to)
 
 	// start sampling from height at checkpoint+1 since the
@@ -226,10 +241,17 @@ func (d *DASer) catchUp(ctx context.Context, job *catchUpJob) (int64, error) {
 		if err != nil {
 			return h.Height - 1, err
 		}
+		d.state.catchUpLk.Lock()
+		d.state.catchUp.Height = uint64(h.Height)
+		d.state.catchUpLk.Unlock()
 	}
+	d.state.catchUpLk.Lock()
+	d.state.catchUp.End = time.Now()
+	d.state.catchUpLk.Unlock()
 
+	jobDetails := d.CatchUpRoutineState()
 	log.Infow("successfully sampled past headers", "from", job.from,
-		"to", job.to, "finished (s)", time.Since(routineStartTime))
+		"to", job.to, "finished (s)", jobDetails.Duration())
 	// report successful result
 	return job.to, nil
 }
@@ -252,7 +274,6 @@ func (d *DASer) sampleHeader(ctx context.Context, h *header.ExtendedHeader) erro
 		}
 		log.Errorw("sampling failed", "height", h.Height, "hash", h.Hash(),
 			"square width", len(h.DAH.RowsRoots), "data root", h.DAH.Hash(), "err", err)
-		log.Warn("DASer WILL BE STOPPED. IN ORDER TO CONTINUE SAMPLING, RE-START THE NODE")
 		// report previous height as the last successfully sampled height
 		return err
 	}
@@ -263,3 +284,52 @@ func (d *DASer) sampleHeader(ctx context.Context, h *header.ExtendedHeader) erro
 
 	return nil
 }
+
+// SampleRoutineState reports the current state of the
+// DASer's main sampling routine.
+func (d *DASer) SampleRoutineState() RoutineState {
+	d.state.sampleLk.RLock()
+	state := d.state.sample
+	d.state.sampleLk.RUnlock()
+	return state
+}
+
+func (d *DASer) updateSampleState(h *header.ExtendedHeader, err error) {
+	height := uint64(h.Height)
+
+	d.state.sampleLk.Lock()
+	defer d.state.sampleLk.Unlock()
+	d.state.sample.LatestSampledHeight = height
+	d.state.sample.LatestSampledSquareWidth = uint64(len(h.DAH.RowsRoots))
+	d.state.sample.Error = err
+}
+
+func (d *DASer) indicateRunning() {
+	d.state.sampleLk.Lock()
+	defer d.state.sampleLk.Unlock()
+	d.state.sample.IsRunning = true
+}
+
+func (d *DASer) indicateStopped() {
+	d.state.sampleLk.Lock()
+	defer d.state.sampleLk.Unlock()
+	d.state.sample.IsRunning = false
+}
+
+// CatchUpRoutineState reports the current state of the
+// DASer's `catchUp` routine.
+func (d *DASer) CatchUpRoutineState() JobInfo {
+	d.state.catchUpLk.RLock()
+	state := d.state.catchUp
+	d.state.catchUpLk.RUnlock()
+	return state
+}
+
+func (d *DASer) recordJobDetails(job *catchUpJob) {
+	d.state.catchUpLk.Lock()
+	defer d.state.catchUpLk.Unlock()
+	d.state.catchUp.ID++
+	d.state.catchUp.From = uint64(job.from)
+	d.state.catchUp.To = uint64(job.to)
+	d.state.catchUp.Start = time.Now()
+}
diff --git a/das/state.go b/das/state.go
new file mode 100644
index 00000000..87d728da
--- /dev/null
+++ b/das/state.go
@@ -0,0 +1,53 @@
+package das
+
+import (
+	"sync"
+	"time"
+)
+
+// state 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.
+type state struct {
+	sampleLk sync.RWMutex
+	sample   RoutineState // tracks information related to the main sampling routine
+
+	catchUpLk sync.RWMutex
+	catchUp   JobInfo // tracks information related to the `catchUp` routine
+}
+
+// RoutineState contains important information about the state of a
+// current sampling routine.
+type RoutineState struct {
+	// reports if an error has occurred during the routine's
+	// sampling process
+	Error error `json:"error"`
+	// tracks the latest successfully sampled height of the routine
+	LatestSampledHeight uint64 `json:"latest_sampled_height"`
+	// tracks the square width of the latest successfully sampled
+	// height of the routine
+	LatestSampledSquareWidth uint64 `json:"latest_sampled_square_width"`
+	// tracks whether routine is running
+	IsRunning bool `json:"is_running"`
+}
+
+// JobInfo contains information about a catchUp job.
+type JobInfo struct {
+	Start time.Time `json:"start"`
+	End   time.Time `json:"end"`
+	Error error     `json:"error"`
+
+	ID     uint64 `json:"id"`
+	Height uint64 `json:"height"`
+	From   uint64 `json:"from"`
+	To     uint64 `json:"to"`
+}
+
+func (ji JobInfo) Finished() bool {
+	return ji.To == ji.Height
+}
+
+func (ji JobInfo) Duration() time.Duration {
+	return ji.End.Sub(ji.Start)
+}
diff --git a/docs/adr/adr-006-fraud-service.md b/docs/adr/adr-006-fraud-service.md
index 0e77d407..d9efa215 100644
--- a/docs/adr/adr-006-fraud-service.md
+++ b/docs/adr/adr-006-fraud-service.md
@@ -12,6 +12,7 @@
   * fixed typo issues;
 - 2022.06.15 - Extend Proof interface with HeaderHash method
 - 2022.06.22 - Updated rsmt2d to change isRow to Axis
+- 2022.07.03 - Add storage description
 
 ## Authors
 
@@ -165,9 +166,13 @@ type Subscription interface {
 ```
 
 ```go
-// FraudSub implements Subscriber and Broadcaster.
-type FraudSub struct {
+// service implements Subscriber and Broadcaster.
+type service struct {
    pubsub *pubsub.PubSub
+   
+   storesLk sync.RWMutex
+   stores   map[ProofType]datastore.Datastore
+   
    topics map[ProofType]*pubsub.Topic
    unmarshallers map[ProofType]ProofUnmarshaller
 }
@@ -188,20 +193,18 @@ Both full and light nodes should stop `DAS`, `Syncer` and `SubmitTx` services.
 
 4. Valid BadEncodingFraudProofs should be stored on the disk using `FraudStore` interface:
 
+### BEFP storage
+BEFP storage will be created on first subscription to Bad Encoding Fraud Proof.
+BEFP will be stored in datastore once it will be received, using `fraud/badEncodingProof` path and the corresponding block hash as the key:
 ```go
-type FraudStore interface {
-   // Put stores given Proof by header's hash in respective for this fraud proof directory
-   Put(path FraudProofType, headerHash string, p Proof) error
-   // Get retrieves Proof by header's hash from respective for this fraud proof directory
-   Get(path FraudProofType, headerHash string) (Proof, error)
-   // GetMany retrieves all Proofs from the respective for this fraud proof directory
-   GetMany(path FraudProofType) ([]Proof, error)
-}
-
-type fraudstore struct{
-   path string
-}
+//  put adds a Fraud Proof to the datastore with the given key.
+func put(ctx context.Context, store datastore.Datastore, key datastore.Key, proof []byte) error
+```
+Once a node starts, it will check if its datastore has a BEFP:
+```go
+func getAll(ctx context.Context, ds datastore.Datastore) ([][]byte, error)
 ```
+In case if response error will be empty (and not ```datastore.ErrNotFound```), then a BEFP has been already added to storage and the node should be halted.
 ### Bridge node behaviour
 Bridge nodes will behave as light nodes do by subscribing to BEFP fraud sub and listening for BEFPs. If a BEFP is received, it will similarly shut down all dependent services, including broadcasting new `ExtendedHeader`s to the network.
 
diff --git a/fraud/bad_encoding.go b/fraud/bad_encoding.go
index 0bf6d8fd..915a5d58 100644
--- a/fraud/bad_encoding.go
+++ b/fraud/bad_encoding.go
@@ -2,7 +2,6 @@ package fraud
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
 
@@ -176,36 +175,3 @@ func (p *BadEncodingProof) Validate(header *header.ExtendedHeader) error {
 
 	return nil
 }
-
-// OnBEFP listens to Bad Encoding Fraud Proof and stops services immediately if it is received.
-func OnBEFP(ctx context.Context, s Subscriber, stop func(context.Context) error) {
-	var err error
-	defer func() {
-		if err == context.Canceled {
-			return
-		}
-
-		stopErr := stop(ctx)
-		if stopErr != nil {
-			log.Warn(stopErr)
-		}
-	}()
-
-	subscription, err := s.Subscribe(BadEncoding)
-	if err != nil {
-		log.Errorw("failed to subscribe to bad encoding fraud proof", "err", err)
-		return
-	}
-	defer subscription.Cancel()
-
-	// At this point we receive already verified fraud proof,
-	// so there are no needs to call Validate.
-	_, err = subscription.Proof(ctx)
-	if err != nil {
-		if err == context.Canceled {
-			return
-		}
-		log.Errorw("reading next proof failed", "err", err)
-		return
-	}
-}
diff --git a/fraud/interface.go b/fraud/interface.go
index 64c0f631..ef448a51 100644
--- a/fraud/interface.go
+++ b/fraud/interface.go
@@ -21,6 +21,7 @@ type ProofUnmarshaler func([]byte) (Proof, error)
 type Service interface {
 	Subscriber
 	Broadcaster
+	Getter
 }
 
 // Broadcaster is a generic interface that sends a `Proof` to all nodes subscribed on the Broadcaster's topic.
@@ -50,3 +51,8 @@ type Subscription interface {
 	Proof(context.Context) (Proof, error)
 	Cancel()
 }
+
+// Getter encompasses the behavior to fetch stored FraudProofs.
+type Getter interface {
+	Get(context.Context, ProofType) ([]Proof, error)
+}
diff --git a/fraud/proof.go b/fraud/proof.go
index 3635bf48..6fda5c93 100644
--- a/fraud/proof.go
+++ b/fraud/proof.go
@@ -7,6 +7,14 @@ import (
 	"github.com/celestiaorg/celestia-node/header"
 )
 
+type ErrFraudExists struct {
+	Proof []Proof
+}
+
+func (e *ErrFraudExists) Error() string {
+	return fmt.Sprintf("fraud: %s proof exists\n", e.Proof[0].Type())
+}
+
 type ProofType int
 
 const (
diff --git a/fraud/service.go b/fraud/service.go
index 07a1b9fd..c4c7ce26 100644
--- a/fraud/service.go
+++ b/fraud/service.go
@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
 	"github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 )
@@ -23,20 +25,25 @@ type service struct {
 	topicsLk sync.RWMutex
 	topics   map[ProofType]*topic
 
+	storesLk sync.RWMutex
+	stores   map[ProofType]datastore.Datastore
+
 	pubsub *pubsub.PubSub
 	getter headerFetcher
+	ds     datastore.Datastore
 }
 
-func NewService(p *pubsub.PubSub, getter headerFetcher) Service {
+func NewService(p *pubsub.PubSub, getter headerFetcher, ds datastore.Datastore) Service {
 	return &service{
 		pubsub: p,
 		getter: getter,
 		topics: make(map[ProofType]*topic),
+		stores: make(map[ProofType]datastore.Datastore),
+		ds:     ds,
 	}
 }
 
 func (f *service) Subscribe(proofType ProofType) (Subscription, error) {
-	// TODO: @vgonkivs check if fraud proof is in fraud store, then return with error
 	f.topicsLk.Lock()
 	t, ok := f.topics[proofType]
 	f.topicsLk.Unlock()
@@ -63,8 +70,6 @@ func (f *service) RegisterUnmarshaler(proofType ProofType, u ProofUnmarshaler) e
 		func(ctx context.Context, _ peer.ID, msg *pubsub.Message) pubsub.ValidationResult {
 			return f.processIncoming(ctx, proofType, msg)
 		},
-		// make validation synchronous.
-		pubsub.WithValidatorInline(true),
 	)
 	if err != nil {
 		return err
@@ -72,6 +77,7 @@ func (f *service) RegisterUnmarshaler(proofType ProofType, u ProofUnmarshaler) e
 	f.topicsLk.Lock()
 	f.topics[proofType] = &topic{topic: t, unmarshal: u}
 	f.topicsLk.Unlock()
+	f.initStore(proofType)
 	return nil
 }
 
@@ -119,8 +125,6 @@ func (f *service) processIncoming(
 		return pubsub.ValidationReject
 	}
 
-	// create a timeout for block fetching since our validator will be called synchronously
-	// and getter is a blocking function.
 	newCtx, cancel := context.WithTimeout(ctx, fetchHeaderTimeout)
 	extHeader, err := f.getter(newCtx, proof.Height())
 	defer cancel()
@@ -147,10 +151,48 @@ func (f *service) processIncoming(
 		"hash", hex.EncodeToString(extHeader.DAH.Hash()),
 		"from", msg.ReceivedFrom.String(),
 	)
+	f.storesLk.RLock()
+	store, ok := f.stores[proofType]
+	f.storesLk.RUnlock()
+	if ok {
+		err = put(ctx, store, string(proof.HeaderHash()), msg.Data)
+		if err != nil {
+			log.Error(err)
+		}
+	} else {
+		log.Warnf("no store for incoming proofs type %s", proof.Type())
+	}
 	log.Warn("Shutting down services...")
 	return pubsub.ValidationAccept
 }
 
+func (f *service) Get(ctx context.Context, proofType ProofType) ([]Proof, error) {
+	f.storesLk.RLock()
+	store, ok := f.stores[proofType]
+	f.storesLk.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("fraud: proof type %s is not supported", proofType)
+	}
+
+	f.topicsLk.RLock()
+	t, ok := f.topics[proofType]
+	f.topicsLk.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("fraud: unmarshaler for proof type %s is not registered", proofType)
+	}
+
+	return getAll(ctx, store, t.unmarshal)
+}
+
+func (f *service) initStore(proofType ProofType) {
+	f.storesLk.Lock()
+	defer f.storesLk.Unlock()
+	_, ok := f.stores[proofType]
+	if !ok {
+		f.stores[proofType] = namespace.Wrap(f.ds, makeKey(proofType))
+	}
+}
+
 func getSubTopic(p ProofType) string {
 	return p.String() + "-sub"
 }
diff --git a/fraud/store.go b/fraud/store.go
new file mode 100644
index 00000000..6c774d84
--- /dev/null
+++ b/fraud/store.go
@@ -0,0 +1,57 @@
+package fraud
+
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	"github.com/ipfs/go-datastore"
+	q "github.com/ipfs/go-datastore/query"
+)
+
+var (
+	storePrefix = "fraud"
+)
+
+// put adds a Fraud Proof to the datastore with the given hash as the key.
+func put(ctx context.Context, ds datastore.Datastore, hash string, value []byte) error {
+	return ds.Put(ctx, datastore.NewKey(hash), value)
+}
+
+// query performs a custom query on the given datastore.
+func query(ctx context.Context, ds datastore.Datastore, q q.Query) ([]q.Entry, error) {
+	results, err := ds.Query(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+
+	return results.Rest()
+}
+
+// getAll queries all Fraud Proofs by their type.
+func getAll(ctx context.Context, ds datastore.Datastore, unmarshal ProofUnmarshaler) ([]Proof, error) {
+	entries, err := query(ctx, ds, q.Query{})
+	if err != nil {
+		return nil, err
+	}
+	if len(entries) == 0 {
+		return nil, datastore.ErrNotFound
+	}
+	proofs := make([]Proof, 0)
+	for _, data := range entries {
+		proof, err := unmarshal(data.Value)
+		if err != nil {
+			log.Warn(err)
+			continue
+		}
+		proofs = append(proofs, proof)
+	}
+	sort.Slice(proofs, func(i, j int) bool {
+		return proofs[i].Height() < proofs[j].Height()
+	})
+	return proofs, nil
+}
+
+func makeKey(p ProofType) datastore.Key {
+	return datastore.NewKey(fmt.Sprintf("%s/%s", storePrefix, p))
+}
diff --git a/fraud/subscription.go b/fraud/subscription.go
index 13f28232..dd76a7bc 100644
--- a/fraud/subscription.go
+++ b/fraud/subscription.go
@@ -33,3 +33,26 @@ func (s *subscription) Proof(ctx context.Context) (Proof, error) {
 func (s *subscription) Cancel() {
 	s.subscription.Cancel()
 }
+
+// OnProof subscribes on a single Fraud Proof.
+// In case a Fraud Proof is received, then the given handle function will be invoked.
+func OnProof(ctx context.Context, subscriber Subscriber, p ProofType, handle func(proof Proof)) {
+	subscription, err := subscriber.Subscribe(p)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+	defer subscription.Cancel()
+
+	// At this point we receive already verified fraud proof,
+	// so there are no needs to call Validate.
+	proof, err := subscription.Proof(ctx)
+	if err != nil {
+		if err != context.Canceled {
+			log.Errorw("reading next proof failed", "err", err)
+		}
+		return
+	}
+
+	handle(proof)
+}
diff --git a/header/sync/sync.go b/header/sync/sync.go
index 1fb64a41..5474bdd6 100644
--- a/header/sync/sync.go
+++ b/header/sync/sync.go
@@ -103,6 +103,11 @@ func (s State) Finished() bool {
 	return s.ToHeight <= s.Height
 }
 
+// Duration returns the duration of the sync.
+func (s State) Duration() time.Duration {
+	return s.End.Sub(s.Start)
+}
+
 // State reports state of the current (if in progress), or last sync (if finished).
 // Note that throughout the whole Syncer lifetime there might an initial sync and multiple catch-ups.
 // All of them are treated as different syncs with different state IDs and other information.
diff --git a/node/components.go b/node/components.go
index 810efe96..71b0fb06 100644
--- a/node/components.go
+++ b/node/components.go
@@ -18,7 +18,10 @@ import (
 	"github.com/celestiaorg/celestia-node/node/services"
 	statecomponents "github.com/celestiaorg/celestia-node/node/state"
 	"github.com/celestiaorg/celestia-node/params"
+	headerServ "github.com/celestiaorg/celestia-node/service/header"
+	rpcServ "github.com/celestiaorg/celestia-node/service/rpc"
 	"github.com/celestiaorg/celestia-node/service/share"
+	"github.com/celestiaorg/celestia-node/service/state"
 )
 
 // lightComponents keeps all the components as DI options required to build a Light Node.
@@ -29,6 +32,7 @@ func lightComponents(cfg *Config, store Store) fx.Option {
 		fx.Provide(services.DASer),
 		fx.Provide(services.HeaderExchangeP2P(cfg.Services)),
 		fx.Provide(services.LightAvailability),
+		fx.Invoke(rpc.Handler),
 	)
 }
 
@@ -40,6 +44,14 @@ func bridgeComponents(cfg *Config, store Store) fx.Option {
 		nodecore.Components(cfg.Core),
 		fx.Supply(header.MakeExtendedHeader),
 		fx.Provide(services.LightAvailability), // TODO(@Wondertan): Remove strict requirements to have Availability
+		fx.Invoke(func(
+			state *state.Service,
+			share *share.Service,
+			header *headerServ.Service,
+			rpcSrv *rpcServ.Server,
+		) {
+			rpc.Handler(state, share, header, rpcSrv, nil)
+		}),
 	)
 }
 
@@ -51,6 +63,7 @@ func fullComponents(cfg *Config, store Store) fx.Option {
 		fx.Provide(services.DASer),
 		fx.Provide(services.HeaderExchangeP2P(cfg.Services)),
 		fx.Provide(services.FullAvailability),
+		fx.Invoke(rpc.Handler),
 	)
 }
 
@@ -79,9 +92,9 @@ func baseComponents(cfg *Config, store Store) fx.Option {
 		fx.Invoke(invokeWatchdog(store.Path())),
 		p2p.Components(cfg.P2P),
 		// state components
-		statecomponents.Components(cfg.Core.GRPCAddr, cfg.Key),
+		statecomponents.Components(cfg.Core.IP, cfg.Core.GRPCPort, cfg.Key),
 		// RPC components
-		rpc.Components(cfg.RPC),
+		fx.Provide(rpc.Server(cfg.RPC)),
 	)
 }
 
diff --git a/node/config_opts.go b/node/config_opts.go
index d91bbe3c..596ba1a6 100644
--- a/node/config_opts.go
+++ b/node/config_opts.go
@@ -1,18 +1,24 @@
 package node
 
-// WithRemoteCore configures Node to start with remote Core.
-func WithRemoteCore(protocol string, address string) Option {
+// WithRemoteCoreIP configures Node to connect to the given remote Core IP.
+func WithRemoteCoreIP(ip string) Option {
 	return func(sets *settings) {
-		sets.cfg.Core.Protocol = protocol
-		sets.cfg.Core.RemoteAddr = address
+		sets.cfg.Core.IP = ip
 	}
 }
 
-// WithGRPCEndpoint configures Node to connect to given gRPC address
+// WithRemoteCorePort configures Node to connect to the given remote Core port.
+func WithRemoteCorePort(port string) Option {
+	return func(sets *settings) {
+		sets.cfg.Core.RPCPort = port
+	}
+}
+
+// WithGRPCPort configures Node to connect to given gRPC port
 // for state-related queries.
-func WithGRPCEndpoint(address string) Option {
+func WithGRPCPort(port string) Option {
 	return func(sets *settings) {
-		sets.cfg.Core.GRPCAddr = address
+		sets.cfg.Core.GRPCPort = port
 	}
 }
 
diff --git a/node/core/core.go b/node/core/core.go
index f797fd60..3d2fcdf9 100644
--- a/node/core/core.go
+++ b/node/core/core.go
@@ -16,12 +16,13 @@ import (
 
 // Config combines all configuration fields for managing the relationship with a Core node.
 type Config struct {
-	Protocol   string
-	RemoteAddr string
-	GRPCAddr   string
+	IP       string
+	RPCPort  string
+	GRPCPort string
 }
 
-// DefaultConfig returns default configuration for Core subsystem.
+// DefaultConfig returns default configuration for managing the
+// node's connection to a Celestia-Core endpoint.
 func DefaultConfig() Config {
 	return Config{}
 }
@@ -33,7 +34,7 @@ func Components(cfg Config) fx.Option {
 		fxutil.ProvideAs(headercore.NewExchange, new(header.Exchange)),
 		fx.Invoke(HeaderListener),
 		fx.Provide(func(lc fx.Lifecycle) (core.Client, error) {
-			if cfg.RemoteAddr == "" {
+			if cfg.IP == "" {
 				return nil, fmt.Errorf("no celestia-core endpoint given")
 			}
 			client, err := RemoteClient(cfg)
@@ -71,5 +72,5 @@ func HeaderListener(
 
 // RemoteClient provides a constructor for core.Client over RPC.
 func RemoteClient(cfg Config) (core.Client, error) {
-	return core.NewRemote(cfg.Protocol, cfg.RemoteAddr)
+	return core.NewRemote(cfg.IP, cfg.RPCPort)
 }
diff --git a/node/rpc/component.go b/node/rpc/component.go
index 721554e1..9c8cc147 100644
--- a/node/rpc/component.go
+++ b/node/rpc/component.go
@@ -3,19 +3,13 @@ package rpc
 import (
 	"go.uber.org/fx"
 
+	"github.com/celestiaorg/celestia-node/das"
 	"github.com/celestiaorg/celestia-node/service/header"
 	"github.com/celestiaorg/celestia-node/service/rpc"
 	"github.com/celestiaorg/celestia-node/service/share"
 	"github.com/celestiaorg/celestia-node/service/state"
 )
 
-func Components(cfg rpc.Config) fx.Option {
-	return fx.Options(
-		fx.Provide(Server(cfg)),
-		fx.Invoke(Handler),
-	)
-}
-
 // Server constructs a new RPC Server from the given Config.
 // TODO @renaynay @Wondertan: this component is meant to be removed on implementation
 //  of https://github.com/celestiaorg/celestia-node/pull/506.
@@ -36,8 +30,9 @@ func Handler(
 	share *share.Service,
 	header *header.Service,
 	serv *rpc.Server,
+	daser *das.DASer,
 ) {
-	handler := rpc.NewHandler(state, share, header)
+	handler := rpc.NewHandler(state, share, header, daser)
 	handler.RegisterEndpoints(serv)
 	handler.RegisterMiddleware(serv)
 }
diff --git a/node/services/service.go b/node/services/service.go
index 770f5f5e..27d8a88f 100644
--- a/node/services/service.go
+++ b/node/services/service.go
@@ -33,14 +33,17 @@ func HeaderSyncer(
 	ex header.Exchange,
 	store header.Store,
 	sub header.Subscriber,
-	fsub fraud.Subscriber,
+	fservice fraud.Service,
 ) (*sync.Syncer, error) {
 	syncer := sync.NewSyncer(ex, store, sub)
+	lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
 	lc.Append(fx.Hook{
-		OnStart: syncer.Start,
-		OnStop:  syncer.Stop,
+		OnStart: func(startCtx context.Context) error {
+			return FraudLifecycle(startCtx, lifecycleCtx, fraud.BadEncoding, fservice, syncer.Start, syncer.Stop)
+		},
+		OnStop: syncer.Stop,
 	})
-	go fraud.OnBEFP(fxutil.WithLifecycle(ctx, lc), fsub, syncer.Stop)
+
 	return syncer, nil
 }
 
@@ -154,21 +157,28 @@ func DASer(
 	fservice fraud.Service,
 ) *das.DASer {
 	das := das.NewDASer(avail, sub, hstore, ds, fservice)
+	lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
 	lc.Append(fx.Hook{
-		OnStart: das.Start,
-		OnStop:  das.Stop,
+		OnStart: func(startContext context.Context) error {
+			return FraudLifecycle(startContext, lifecycleCtx, fraud.BadEncoding, fservice, das.Start, das.Stop)
+		},
+		OnStop: das.Stop,
 	})
-	go fraud.OnBEFP(fxutil.WithLifecycle(ctx, lc), fservice, das.Stop)
+
 	return das
 }
 
-// FraudService constructs fraud proof service for bad encoding fraud proofs (BEFPs).
-func FraudService(sub *pubsub.PubSub, hstore header.Store) (fraud.Service, fraud.Service, error) {
-	f := fraud.NewService(sub, hstore.GetByHeight)
+// FraudService constructs fraud proof service.
+func FraudService(
+	sub *pubsub.PubSub,
+	hstore header.Store,
+	ds datastore.Batching,
+) (fraud.Service, error) {
+	f := fraud.NewService(sub, hstore.GetByHeight, ds)
 	if err := f.RegisterUnmarshaler(fraud.BadEncoding, fraud.UnmarshalBEFP); err != nil {
-		return nil, nil, err
+		return nil, err
 	}
-	return f, f, nil
+	return f, nil
 }
 
 // LightAvailability constructs light share availability wrapped in cache availability.
@@ -212,3 +222,33 @@ func FullAvailability(
 	})
 	return ca
 }
+
+// FraudLifecycle controls the lifecycle of service depending on fraud proofs.
+// It starts the service only if no fraud-proof exists and stops the service automatically
+// if a proof arrives after the service was started.
+func FraudLifecycle(
+	startCtx, lifecycleCtx context.Context,
+	p fraud.ProofType,
+	fservice fraud.Service,
+	start, stop func(context.Context) error,
+) error {
+	proofs, err := fservice.Get(startCtx, p)
+	switch err {
+	default:
+		return err
+	case nil:
+		return &fraud.ErrFraudExists{Proof: proofs}
+	case datastore.ErrNotFound:
+	}
+	err = start(startCtx)
+	if err != nil {
+		return err
+	}
+	// handle incoming Fraud Proofs
+	go fraud.OnProof(lifecycleCtx, fservice, p, func(fraud.Proof) {
+		if err := stop(lifecycleCtx); err != nil {
+			log.Error(err)
+		}
+	})
+	return nil
+}
diff --git a/node/state/core.go b/node/state/core.go
index afe6da25..0484f1df 100644
--- a/node/state/core.go
+++ b/node/state/core.go
@@ -10,10 +10,11 @@ import (
 // CoreAccessor constructs a new instance of state.Accessor over
 // a celestia-core connection.
 func CoreAccessor(
-	endpoint string,
+	coreIP,
+	grpcPort string,
 ) func(fx.Lifecycle, *apptypes.KeyringSigner) (state.Accessor, error) {
 	return func(lc fx.Lifecycle, signer *apptypes.KeyringSigner) (state.Accessor, error) {
-		ca := state.NewCoreAccessor(signer, endpoint)
+		ca := state.NewCoreAccessor(signer, coreIP, grpcPort)
 		lc.Append(fx.Hook{
 			OnStart: ca.Start,
 			OnStop:  ca.Stop,
diff --git a/node/state/state.go b/node/state/state.go
index ac4b70ee..67ff64ee 100644
--- a/node/state/state.go
+++ b/node/state/state.go
@@ -9,6 +9,7 @@ import (
 	"github.com/celestiaorg/celestia-node/fraud"
 	"github.com/celestiaorg/celestia-node/libs/fxutil"
 	"github.com/celestiaorg/celestia-node/node/key"
+	"github.com/celestiaorg/celestia-node/node/services"
 	"github.com/celestiaorg/celestia-node/service/state"
 )
 
@@ -16,21 +17,24 @@ var log = logging.Logger("state-access-constructor")
 
 // Components provides all components necessary to construct the
 // state service.
-func Components(coreEndpoint string, cfg key.Config) fx.Option {
+func Components(coreIP, grpcPort string, cfg key.Config) fx.Option {
 	return fx.Options(
 		fx.Provide(Keyring(cfg)),
-		fx.Provide(CoreAccessor(coreEndpoint)),
+		fx.Provide(CoreAccessor(coreIP, grpcPort)),
 		fx.Provide(Service),
 	)
 }
 
 // Service constructs a new state.Service.
-func Service(ctx context.Context, lc fx.Lifecycle, accessor state.Accessor, fsub fraud.Subscriber) *state.Service {
+func Service(ctx context.Context, lc fx.Lifecycle, accessor state.Accessor, fservice fraud.Service) *state.Service {
 	serv := state.NewService(accessor)
+	lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
 	lc.Append(fx.Hook{
-		OnStart: serv.Start,
-		OnStop:  serv.Stop,
+		OnStart: func(startCtx context.Context) error {
+			return services.FraudLifecycle(startCtx, lifecycleCtx, fraud.BadEncoding, fservice, serv.Start, serv.Stop)
+		},
+		OnStop: serv.Stop,
 	})
-	go fraud.OnBEFP(fxutil.WithLifecycle(ctx, lc), fsub, serv.Stop)
+
 	return serv
 }
diff --git a/service/rpc/das.go b/service/rpc/das.go
new file mode 100644
index 00000000..f6369a14
--- /dev/null
+++ b/service/rpc/das.go
@@ -0,0 +1,35 @@
+package rpc
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/celestiaorg/celestia-node/das"
+)
+
+const (
+	dasStateEndpoint = "/daser/state"
+)
+
+// DasStateResponse encompasses the fields returned in response
+// to a `/daser` request.
+type DasStateResponse struct {
+	SampleRoutine  das.RoutineState `json:"sample_routine"`
+	CatchUpRoutine das.JobInfo      `json:"catch_up_routine"`
+}
+
+func (h *Handler) handleDASStateRequest(w http.ResponseWriter, r *http.Request) {
+	dasState := new(DasStateResponse)
+	dasState.SampleRoutine = h.das.SampleRoutineState()
+	dasState.CatchUpRoutine = h.das.CatchUpRoutineState()
+
+	resp, err := json.Marshal(dasState)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, dasStateEndpoint, err)
+		return
+	}
+	_, err = w.Write(resp)
+	if err != nil {
+		log.Errorw("serving request", "endpoint", dasStateEndpoint, "err", err)
+	}
+}
diff --git a/service/rpc/endpoints.go b/service/rpc/endpoints.go
index 6a732bf6..62b4764a 100644
--- a/service/rpc/endpoints.go
+++ b/service/rpc/endpoints.go
@@ -32,4 +32,10 @@ func (h *Handler) RegisterEndpoints(rpc *Server) {
 	rpc.RegisterHandlerFunc(fmt.Sprintf("%s/{%s}", headerByHeightEndpoint, heightKey), h.handleHeaderRequest,
 		http.MethodGet)
 	rpc.RegisterHandlerFunc(headEndpoint, h.handleHeadRequest, http.MethodGet)
+
+	// DASer endpoints
+	// only register if DASer service is available
+	if h.das != nil {
+		rpc.RegisterHandlerFunc(dasStateEndpoint, h.handleDASStateRequest, http.MethodGet)
+	}
 }
diff --git a/service/rpc/handler.go b/service/rpc/handler.go
index 88efffd5..fc3d9406 100644
--- a/service/rpc/handler.go
+++ b/service/rpc/handler.go
@@ -3,6 +3,7 @@ package rpc
 import (
 	logging "github.com/ipfs/go-log/v2"
 
+	"github.com/celestiaorg/celestia-node/das"
 	"github.com/celestiaorg/celestia-node/service/header"
 	"github.com/celestiaorg/celestia-node/service/share"
 	"github.com/celestiaorg/celestia-node/service/state"
@@ -14,12 +15,19 @@ type Handler struct {
 	state  *state.Service
 	share  *share.Service
 	header *header.Service
+	das    *das.DASer
 }
 
-func NewHandler(state *state.Service, share *share.Service, header *header.Service) *Handler {
+func NewHandler(
+	state *state.Service,
+	share *share.Service,
+	header *header.Service,
+	das *das.DASer,
+) *Handler {
 	return &Handler{
 		state:  state,
 		share:  share,
 		header: header,
+		das:    das,
 	}
 }
diff --git a/service/state/core_access.go b/service/state/core_access.go
index d47e4698..ff8123fb 100644
--- a/service/state/core_access.go
+++ b/service/state/core_access.go
@@ -21,9 +21,10 @@ import (
 type CoreAccessor struct {
 	signer *apptypes.KeyringSigner
 
-	coreEndpoint string
-	coreConn     *grpc.ClientConn
-	queryCli     banktypes.QueryClient
+	coreIP   string
+	grpcPort string
+	coreConn *grpc.ClientConn
+	queryCli banktypes.QueryClient
 }
 
 // NewCoreAccessor dials the given celestia-core endpoint and
@@ -31,11 +32,13 @@ type CoreAccessor struct {
 // connection.
 func NewCoreAccessor(
 	signer *apptypes.KeyringSigner,
-	endpoint string,
+	coreIP,
+	grpcPort string,
 ) *CoreAccessor {
 	return &CoreAccessor{
-		signer:       signer,
-		coreEndpoint: endpoint,
+		signer:   signer,
+		coreIP:   coreIP,
+		grpcPort: grpcPort,
 	}
 }
 
@@ -44,7 +47,8 @@ func (ca *CoreAccessor) Start(ctx context.Context) error {
 		return fmt.Errorf("core-access: already connected to core endpoint")
 	}
 	// dial given celestia-core endpoint
-	client, err := grpc.DialContext(ctx, ca.coreEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	endpoint := fmt.Sprintf("%s:%s", ca.coreIP, ca.grpcPort)
+	client, err := grpc.DialContext(ctx, endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		return err
 	}


Test patch:
diff --git a/core/testing.go b/core/testing.go
index b74a597c..8b45ecc2 100644
--- a/core/testing.go
+++ b/core/testing.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
+	"net"
+	"net/url"
 	"sort"
 	"testing"
 	"time"
@@ -51,8 +53,11 @@ func CreateKVStore(retainBlocks int64) *kvstore.Application {
 // mock Core Client.
 func StartTestClient(ctx context.Context, t *testing.T) (tmservice.Service, Client) {
 	nd, _, cfg := StartTestKVApp(ctx, t)
-	protocol, ip := GetEndpoint(cfg)
-	client, err := NewRemote(protocol, ip)
+	endpoint, err := GetEndpoint(cfg)
+	require.NoError(t, err)
+	ip, port, err := net.SplitHostPort(endpoint)
+	require.NoError(t, err)
+	client, err := NewRemote(ip, port)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		err := client.Stop()
@@ -63,11 +68,17 @@ func StartTestClient(ctx context.Context, t *testing.T) (tmservice.Service, Clie
 	return nd, client
 }
 
-// GetEndpoint returns the protocol and ip of the remote node.
-func GetEndpoint(cfg *config.Config) (string, string) {
-	endpoint := cfg.RPC.ListenAddress
-	protocol, ip := endpoint[:3], endpoint[6:]
-	return protocol, ip
+// GetEndpoint returns the remote node's RPC endpoint.
+func GetEndpoint(cfg *config.Config) (string, error) {
+	url, err := url.Parse(cfg.RPC.ListenAddress)
+	if err != nil {
+		return "", err
+	}
+	host, _, err := net.SplitHostPort(url.Host)
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s:%s", host, url.Port()), nil
 }
 
 func RandValidator(randPower bool, minPower int64) (*tmtypes.Validator, tmtypes.PrivValidator) {
diff --git a/das/daser_test.go b/das/daser_test.go
index 5fc6ad29..81f1d3f0 100644
--- a/das/daser_test.go
+++ b/das/daser_test.go
@@ -39,28 +39,33 @@ func TestDASerLifecycle(t *testing.T) {
 
 	err := daser.Start(ctx)
 	require.NoError(t, err)
-
+	defer func() {
+		err = daser.Stop(ctx)
+		require.NoError(t, err)
+
+		// load checkpoint and ensure it's at network head
+		checkpoint, err := loadCheckpoint(ctx, daser.cstore)
+		require.NoError(t, err)
+		// ensure checkpoint is stored at 15
+		assert.Equal(t, int64(15), checkpoint)
+	}()
 	// wait for dasing catch-up routine to finish
 	select {
 	case <-ctx.Done():
 		t.Fatal(ctx.Err())
 	case <-mockGet.doneCh:
 	}
-
 	// give catch-up routine a second to finish up sampling last header
-	// TODO @renaynay: this sleep is a known flakey solution to the issue that
-	//  we do not have DASState implemented yet. Once DASState is implemented, rely
-	//  on that instead of sleeping.
-	time.Sleep(time.Second * 1)
-
-	err = daser.Stop(ctx)
-	require.NoError(t, err)
-
-	// load checkpoint and ensure it's at network head
-	checkpoint, err := loadCheckpoint(ctx, daser.cstore)
-	require.NoError(t, err)
-	// ensure checkpoint is stored at 15
-	assert.Equal(t, int64(15), checkpoint)
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal(ctx.Err())
+		default:
+			if daser.CatchUpRoutineState().Finished() {
+				return
+			}
+		}
+	}
 }
 
 func TestDASer_Restart(t *testing.T) {
@@ -106,19 +111,15 @@ func TestDASer_Restart(t *testing.T) {
 	// wait for dasing catch-up routine to finish
 	select {
 	case <-restartCtx.Done():
-		t.Fatal(ctx.Err())
+		t.Fatal(restartCtx.Err())
 	case <-mockGet.doneCh:
 	}
 
-	// give catch-up routine a second to finish up sampling last header
-	// TODO @renaynay: this sleep is a known flakey solution to the issue that
-	//  we do not have DASState implemented yet. Once DASState is implemented, rely
-	//  on that instead of sleeping.
-	time.Sleep(time.Second * 1)
-
 	err = daser.Stop(restartCtx)
 	require.NoError(t, err)
 
+	assert.True(t, daser.CatchUpRoutineState().Finished())
+
 	// load checkpoint and ensure it's at network head
 	checkpoint, err := loadCheckpoint(ctx, daser.cstore)
 	require.NoError(t, err)
@@ -215,7 +216,7 @@ func TestDASer_catchUp_fails(t *testing.T) {
 	bServ := mdutils.Bserv()
 	avail := share.TestLightAvailability(bServ)
 	mockGet, _, _, mockService := createDASerSubcomponents(t, bServ, 6, 0, avail)
-	daser := NewDASer(share.NewBrokenAvailability(), nil, mockGet, ds, mockService)
+	daser := NewDASer(share.NewTestBrokenAvailability(), nil, mockGet, ds, mockService)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
@@ -271,30 +272,126 @@ func TestDASer_stopsAfter_BEFP(t *testing.T) {
 	mockGet, shareServ, sub, _ := createDASerSubcomponents(t, bServ, 15, 15, avail)
 
 	// create fraud service and break one header
-	f := fraud.NewService(ps, mockGet.GetByHeight)
+	f := fraud.NewService(ps, mockGet.GetByHeight, ds)
 	require.NoError(t, f.RegisterUnmarshaler(fraud.BadEncoding, fraud.UnmarshalBEFP))
 	mockGet.headers[1] = header.CreateFraudExtHeader(t, mockGet.headers[1], bServ)
 
 	// create and start DASer
 	daser := NewDASer(shareServ, sub, mockGet, ds, f)
-	go fraud.OnBEFP(ctx, f, daser.Stop)
-	require.NoError(t, daser.Start(ctx))
 
+	require.NoError(t, daser.Start(ctx))
+	fraud.OnProof(ctx, f, fraud.BadEncoding,
+		func(proof fraud.Proof) {
+			require.NoError(t, daser.Stop(ctx))
+		})
 	// wait for dasing catch-up routine fails
 	select {
 	case <-ctx.Done():
 		t.Fatal(ctx.Err())
 	case <-daser.ctx.Done():
-		break
 	}
 
-	// give catch-up routine a second to finish up sampling last header
-	// TODO @renaynay: this sleep is a known flakey solution to the issue that
-	//  we do not have DASState implemented yet. Once DASState is implemented, rely
-	//  on that instead of sleeping.
-	time.Sleep(time.Second * 1)
 	require.True(t, daser.ctx.Err() == context.Canceled)
 }
+func TestDASerState(t *testing.T) {
+	ds := ds_sync.MutexWrap(datastore.NewMapDatastore())
+	bServ := mdutils.Bserv()
+
+	// 30 headers in the past and 30 headers in the future (so final sample height should be 60)
+	mockGet, shareServ, sub, fraud := createDASerSubcomponents(t, bServ, 30, 30, share.NewTestSuccessfulAvailability())
+	expectedFinalSampleHeight := uint64(60)
+	expectedFinalSampleWidth := uint64(len(sub.Headers[29].DAH.RowsRoots))
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	t.Cleanup(cancel)
+
+	daser := NewDASer(shareServ, sub, mockGet, ds, fraud)
+	err := daser.Start(ctx)
+	require.NoError(t, err)
+	defer func() {
+		// wait for all "future" headers to be sampled
+		for {
+			select {
+			case <-ctx.Done():
+				t.Fatal(ctx.Err())
+			default:
+				if daser.SampleRoutineState().LatestSampledHeight == expectedFinalSampleHeight {
+					assert.Equal(t, expectedFinalSampleWidth, daser.SampleRoutineState().LatestSampledSquareWidth)
+
+					err := daser.Stop(ctx)
+					require.NoError(t, err)
+					assert.False(t, daser.SampleRoutineState().IsRunning)
+					return
+				}
+			}
+		}
+	}()
+	select {
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	case <-mockGet.doneCh:
+	}
+	// give catchUp routine a second to exit
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal(ctx.Err())
+		default:
+			if daser.CatchUpRoutineState().Finished() {
+				return
+			}
+		}
+	}
+}
+
+// TestDASerState_WithErrorInCatchUp tests for the case where an
+// error has occurred inside the catchUp routine, ensuring the catchUp
+// routine exits as expected and reports the error accurately.
+func TestDASerState_WithErrorInCatchUp(t *testing.T) {
+	ds := ds_sync.MutexWrap(datastore.NewMapDatastore())
+	bServ := mdutils.Bserv()
+
+	// 30 headers in the past and 30 headers in the future
+	// catchUp routine error should occur at height 16
+	brokenHeight := 16
+	mockGet, shareServ, sub, fraud := createDASerWithFailingAvailability(t, bServ, 30, 30,
+		int64(brokenHeight))
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	t.Cleanup(cancel)
+
+	daser := NewDASer(shareServ, sub, mockGet, ds, fraud)
+	// allow catchUpDn signal to be read twice
+	daser.catchUpDn = make(chan struct{}, 2)
+
+	err := daser.Start(ctx)
+	require.NoError(t, err)
+	defer func() {
+		err = daser.Stop(ctx)
+		require.NoError(t, err)
+	}()
+	// wait until catchUp routine has fetched the broken header
+	select {
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	case <-mockGet.brokenHeightCh:
+	}
+	// wait for daser to exit catchUp routine
+	select {
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	case <-daser.catchUpDn:
+		catchUp := daser.CatchUpRoutineState()
+		// make sure catchUp routine didn't successfully complete
+		assert.False(t, catchUp.Finished())
+		// ensure that the error has been reported
+		assert.NotNil(t, catchUp.Error)
+		assert.Equal(t, uint64(brokenHeight)-1, catchUp.Height)
+		// send another done signal so `Stop` can read it
+		daser.catchUpDn <- struct{}{}
+		return
+	}
+}
 
 // createDASerSubcomponents takes numGetter (number of headers
 // to store in mockGetter) and numSub (number of headers to store
@@ -308,19 +405,46 @@ func createDASerSubcomponents(
 	availability share.Availability,
 ) (*mockGetter, *share.Service, *header.DummySubscriber, *fraud.DummyService) {
 	shareServ := share.NewService(bServ, availability)
+	mockGet, sub := createMockGetterAndSub(t, bServ, numGetter, numSub)
+	fraud := new(fraud.DummyService)
+	return mockGet, shareServ, sub, fraud
+}
 
+func createMockGetterAndSub(
+	t *testing.T,
+	bServ blockservice.BlockService,
+	numGetter,
+	numSub int,
+) (*mockGetter, *header.DummySubscriber) {
 	mockGet := &mockGetter{
-		headers: make(map[int64]*header.ExtendedHeader),
-		doneCh:  make(chan struct{}),
+		headers:        make(map[int64]*header.ExtendedHeader),
+		doneCh:         make(chan struct{}),
+		brokenHeightCh: make(chan struct{}),
 	}
 
 	mockGet.generateHeaders(t, bServ, 0, numGetter)
 
 	sub := new(header.DummySubscriber)
-	fraud := new(fraud.DummyService)
 	mockGet.fillSubWithHeaders(t, sub, bServ, numGetter, numGetter+numSub)
 
-	return mockGet, shareServ, sub, fraud
+	return mockGet, sub
+}
+
+func createDASerWithFailingAvailability(
+	t *testing.T,
+	bServ blockservice.BlockService,
+	numGetter,
+	numSub int,
+	brokenHeight int64,
+) (*mockGetter, *share.Service, *header.DummySubscriber, *fraud.DummyService) {
+	mockGet, sub := createMockGetterAndSub(t, bServ, numGetter, numSub)
+	mockGet.brokenHeight = brokenHeight
+
+	shareServ := share.NewService(bServ, &share.TestBrokenAvailability{
+		Root: mockGet.headers[brokenHeight].DAH,
+	})
+
+	return mockGet, shareServ, sub, new(fraud.DummyService)
 }
 
 // fillSubWithHeaders generates `num` headers from the future for p2pSub to pipe through to DASer.
@@ -353,6 +477,9 @@ func (m *mockGetter) fillSubWithHeaders(
 type mockGetter struct {
 	doneCh chan struct{} // signals all stored headers have been retrieved
 
+	brokenHeight   int64
+	brokenHeightCh chan struct{}
+
 	head    int64
 	headers map[int64]*header.ExtendedHeader
 }
@@ -378,10 +505,14 @@ func (m *mockGetter) Head(context.Context) (*header.ExtendedHeader, error) {
 
 func (m *mockGetter) GetByHeight(_ context.Context, height uint64) (*header.ExtendedHeader, error) {
 	defer func() {
-		if int64(height) == m.head {
+		switch int64(height) {
+		case m.brokenHeight:
+			close(m.brokenHeightCh)
+		case m.head:
 			close(m.doneCh)
 		}
 	}()
+
 	return m.headers[int64(height)], nil
 }
 
diff --git a/fraud/service_test.go b/fraud/service_test.go
index 9ad84647..c8738ffb 100644
--- a/fraud/service_test.go
+++ b/fraud/service_test.go
@@ -8,6 +8,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/sync"
 	mdutils "github.com/ipfs/go-merkledag/test"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
@@ -91,5 +93,5 @@ func createService(t *testing.T) (Service, *mockStore) {
 		pubsub.WithMessageSignaturePolicy(pubsub.StrictNoSign))
 	require.NoError(t, err)
 	store := createStore(t, 10)
-	return NewService(ps, store.GetByHeight), store
+	return NewService(ps, store.GetByHeight, sync.MutexWrap(datastore.NewMapDatastore())), store
 }
diff --git a/fraud/store_test.go b/fraud/store_test.go
new file mode 100644
index 00000000..2bffa299
--- /dev/null
+++ b/fraud/store_test.go
@@ -0,0 +1,81 @@
+package fraud
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
+	ds_sync "github.com/ipfs/go-datastore/sync"
+	mdutils "github.com/ipfs/go-merkledag/test"
+	"github.com/stretchr/testify/require"
+
+	"github.com/celestiaorg/celestia-node/header"
+	"github.com/celestiaorg/celestia-node/ipld"
+)
+
+func TestStore_Put(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+	defer t.Cleanup(cancel)
+	bServ := mdutils.Bserv()
+	_, store := createService(t)
+	h, err := store.GetByHeight(ctx, 1)
+	require.NoError(t, err)
+
+	faultDAH, err := generateByzantineError(ctx, t, h, bServ)
+	var errByz *ipld.ErrByzantine
+	require.True(t, errors.As(err, &errByz))
+	p := CreateBadEncodingProof([]byte("hash"), uint64(faultDAH.Height), errByz)
+	bin, err := p.MarshalBinary()
+	require.NoError(t, err)
+	ds := ds_sync.MutexWrap(datastore.NewMapDatastore())
+	badEncodingStore := namespace.Wrap(ds, makeKey(BadEncoding))
+	err = put(ctx, badEncodingStore, string(p.HeaderHash()), bin)
+	require.NoError(t, err)
+}
+
+func TestStore_GetAll(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+	defer t.Cleanup(cancel)
+	bServ := mdutils.Bserv()
+	_, store := createService(t)
+
+	ds := ds_sync.MutexWrap(datastore.NewMapDatastore())
+	badEncodingStore := namespace.Wrap(ds, makeKey(BadEncoding))
+	faultHeaders := make([]*header.ExtendedHeader, 0)
+	for i := 0; i < 3; i++ {
+		h, err := store.GetByHeight(ctx, uint64(i+1))
+		require.NoError(t, err)
+		faultDAH, err := generateByzantineError(ctx, t, h, bServ)
+		var errByz *ipld.ErrByzantine
+		require.True(t, errors.As(err, &errByz))
+
+		p := CreateBadEncodingProof(h.Hash(), uint64(faultDAH.Height), errByz)
+		bin, err := p.MarshalBinary()
+		require.NoError(t, err)
+		err = put(ctx, badEncodingStore, string(p.HeaderHash()), bin)
+		require.NoError(t, err)
+		faultHeaders = append(faultHeaders, faultDAH)
+	}
+	befp, err := getAll(ctx, badEncodingStore, UnmarshalBEFP)
+	require.NoError(t, err)
+	require.NotEmpty(t, befp)
+	for i := 0; i < len(befp); i++ {
+		require.NoError(t, befp[i].Validate(faultHeaders[i]))
+	}
+}
+
+func Test_GetAllFailed(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+	defer t.Cleanup(cancel)
+
+	ds := ds_sync.MutexWrap(datastore.NewMapDatastore())
+	badEncodingStore := namespace.Wrap(ds, makeKey(BadEncoding))
+
+	proofs, err := getAll(ctx, badEncodingStore, UnmarshalBEFP)
+	require.Error(t, err)
+	require.ErrorIs(t, err, datastore.ErrNotFound)
+	require.Nil(t, proofs)
+}
diff --git a/fraud/testing.go b/fraud/testing.go
index b9b1c4f4..5006cd28 100644
--- a/fraud/testing.go
+++ b/fraud/testing.go
@@ -37,6 +37,10 @@ func (d *DummyService) UnregisterUnmarshaler(ProofType) error {
 	return nil
 }
 
+func (d *DummyService) Get(context.Context, ProofType) ([]Proof, error) {
+	return nil, nil
+}
+
 type dummySubscription struct {
 }
 
diff --git a/node/rpc_test.go b/node/rpc_test.go
index de312e11..7a8b78fa 100644
--- a/node/rpc_test.go
+++ b/node/rpc_test.go
@@ -9,16 +9,20 @@ import (
 	"strconv"
 	"testing"
 
+	"github.com/ipfs/go-datastore"
+	ds_sync "github.com/ipfs/go-datastore/sync"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/fx"
 
+	"github.com/celestiaorg/celestia-node/das"
 	"github.com/celestiaorg/celestia-node/header"
 	"github.com/celestiaorg/celestia-node/header/local"
 	"github.com/celestiaorg/celestia-node/header/store"
 	"github.com/celestiaorg/celestia-node/header/sync"
 	service "github.com/celestiaorg/celestia-node/service/header"
 	"github.com/celestiaorg/celestia-node/service/rpc"
+	"github.com/celestiaorg/celestia-node/service/share"
 )
 
 // NOTE: The following tests are against common RPC endpoints provided by
@@ -173,24 +177,52 @@ func TestAvailabilityRequest(t *testing.T) {
 	assert.True(t, availResp.Available)
 }
 
+func TestDASStateRequest(t *testing.T) {
+	nd := setupNodeWithModifiedRPC(t)
+
+	endpoint := fmt.Sprintf("http://127.0.0.1:%s/daser/state", nd.RPCServer.ListenAddr()[5:])
+	resp, err := http.Get(endpoint)
+	require.NoError(t, err)
+	defer func() {
+		err = resp.Body.Close()
+		require.NoError(t, err)
+	}()
+	dasStateResp := new(rpc.DasStateResponse)
+	err = json.NewDecoder(resp.Body).Decode(dasStateResp)
+	require.NoError(t, err)
+	// ensure daser is running
+	assert.True(t, dasStateResp.SampleRoutine.IsRunning)
+}
+
 func setupNodeWithModifiedRPC(t *testing.T) *Node {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 	// create test node with a dummy header service, manually add a dummy header
 	// service and register it with rpc handler/server
 	hServ := setupHeaderService(ctx, t)
+	daser := setupDASer()
 	// create overrides
 	overrideHeaderServ := func(sets *settings) {
 		sets.opts = append(sets.opts, fx.Replace(hServ))
 	}
+	overrideDASer := func(sets *settings) {
+		sets.opts = append(sets.opts, fx.Replace(func() func(lc fx.Lifecycle) *das.DASer {
+			return func(lc fx.Lifecycle) *das.DASer {
+				lc.Append(fx.Hook{
+					OnStart: daser.Start,
+					OnStop:  daser.Stop,
+				})
+				return daser
+			}
+		}))
+	}
 	overrideRPCHandler := func(sets *settings) {
-		sets.opts = append(sets.opts, fx.Replace(func(srv *rpc.Server) *rpc.Handler {
-			handler := rpc.NewHandler(nil, nil, hServ)
+		sets.opts = append(sets.opts, fx.Invoke(func(srv *rpc.Server) {
+			handler := rpc.NewHandler(nil, nil, hServ, daser)
 			handler.RegisterEndpoints(srv)
-			return handler
 		}))
 	}
-	nd := TestNode(t, Full, overrideHeaderServ, overrideRPCHandler)
+	nd := TestNode(t, Full, overrideHeaderServ, overrideDASer, overrideRPCHandler)
 	// start node
 	err := nd.Start(ctx)
 	require.NoError(t, err)
@@ -214,3 +246,9 @@ func setupHeaderService(ctx context.Context, t *testing.T) *service.Service {
 
 	return service.NewHeaderService(syncer, nil, nil, nil, localStore)
 }
+
+func setupDASer() *das.DASer {
+	ds := ds_sync.MutexWrap(datastore.NewMapDatastore())
+	sub := &header.DummySubscriber{Headers: make([]*header.ExtendedHeader, 10)}
+	return das.NewDASer(share.NewTestSuccessfulAvailability(), sub, nil, ds, nil)
+}
diff --git a/node/testing.go b/node/testing.go
index 79b56c6b..2961402c 100644
--- a/node/testing.go
+++ b/node/testing.go
@@ -2,6 +2,7 @@ package node
 
 import (
 	"context"
+	"net"
 	"testing"
 	"time"
 
@@ -31,8 +32,13 @@ func TestNode(t *testing.T, tp Type, opts ...Option) *Node {
 
 	store := MockStore(t, DefaultConfig(tp))
 	_, _, cfg := core.StartTestKVApp(ctx, t)
+	endpoint, err := core.GetEndpoint(cfg)
+	require.NoError(t, err)
+	ip, port, err := net.SplitHostPort(endpoint)
+	require.NoError(t, err)
 	opts = append(opts,
-		WithRemoteCore(core.GetEndpoint(cfg)),
+		WithRemoteCoreIP(ip),
+		WithRemoteCorePort(port),
 		WithNetwork(params.Private),
 		WithRPCPort("0"),
 		WithKeyringSigner(TestKeyringSigner(t)),
diff --git a/node/tests/fraud_test.go b/node/tests/fraud_test.go
index 60cbbfd1..e7d355bc 100644
--- a/node/tests/fraud_test.go
+++ b/node/tests/fraud_test.go
@@ -40,7 +40,9 @@ func TestFraudProofBroadcasting(t *testing.T) {
 	require.NoError(t, err)
 	addrs, err := peer.AddrInfoToP2pAddrs(host.InfoFromHost(bridge.Host))
 	require.NoError(t, err)
-	full := sw.NewFullNode(node.WithTrustedPeers(addrs[0].String()))
+
+	store := node.MockStore(t, node.DefaultConfig(node.Full))
+	full := sw.NewNodeWithStore(node.Full, store, node.WithTrustedPeers(addrs[0].String()))
 
 	err = full.Start(ctx)
 	require.NoError(t, err)
@@ -58,4 +60,13 @@ func TestFraudProofBroadcasting(t *testing.T) {
 
 	_, err = full.HeaderServ.GetByHeight(newCtx, 15)
 	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	require.NoError(t, full.Stop(ctx))
+	require.NoError(t, sw.RemoveNode(full, node.Full))
+
+	full = sw.NewNodeWithStore(node.Full, store, node.WithTrustedPeers(addrs[0].String()))
+	require.Error(t, full.Start(ctx))
+	proofs, err := full.FraudServ.Get(ctx, fraud.BadEncoding)
+	require.NoError(t, err)
+	require.NotNil(t, proofs)
 }
diff --git a/node/tests/swamp/swamp.go b/node/tests/swamp/swamp.go
index 72e86d36..f316564b 100644
--- a/node/tests/swamp/swamp.go
+++ b/node/tests/swamp/swamp.go
@@ -63,8 +63,11 @@ func NewSwamp(t *testing.T, options ...Option) *Swamp {
 	// so, we are not creating bridge nodes with each one containing its own core client
 	// instead we are assigning all created BNs to 1 Core from the swamp
 	core.StartTestNode(ctx, t, ic.App, ic.CoreCfg)
-	protocol, ip := core.GetEndpoint(ic.CoreCfg)
-	remote, err := core.NewRemote(protocol, ip)
+	endpoint, err := core.GetEndpoint(ic.CoreCfg)
+	require.NoError(t, err)
+	ip, port, err := net.SplitHostPort(endpoint)
+	require.NoError(t, err)
+	remote, err := core.NewRemote(ip, port)
 	require.NoError(t, err)
 
 	err = remote.Start()
diff --git a/service/share/testing.go b/service/share/testing.go
index 4882d9bc..016d0e34 100644
--- a/service/share/testing.go
+++ b/service/share/testing.go
@@ -1,6 +1,7 @@
 package share
 
 import (
+	"bytes"
 	"context"
 	"testing"
 
@@ -240,18 +241,28 @@ func (sn *subNet) ConnectAll() {
 	}
 }
 
-// brokenAvailability allows to test error cases during sampling
-type brokenAvailability struct{}
+type TestBrokenAvailability struct {
+	Root *Root
+}
 
-func NewBrokenAvailability() Availability {
-	return &brokenAvailability{}
+// NewTestBrokenAvailability returns an instance of Availability that
+// allows for testing error cases during sampling.
+//
+// If the Root field is empty, it will return ErrNotAvailable on every call
+// to SharesAvailable. Otherwise, it will only return ErrNotAvailable if the
+// given Root hash matches the stored Root hash.
+func NewTestBrokenAvailability() Availability {
+	return &TestBrokenAvailability{}
 }
 
-func (b *brokenAvailability) SharesAvailable(context.Context, *Root) error {
-	return ErrNotAvailable
+func (b *TestBrokenAvailability) SharesAvailable(_ context.Context, root *Root) error {
+	if b.Root == nil || bytes.Equal(b.Root.Hash(), root.Hash()) {
+		return ErrNotAvailable
+	}
+	return nil
 }
 
-func (b *brokenAvailability) ProbabilityOfAvailability() float64 {
+func (b *TestBrokenAvailability) ProbabilityOfAvailability() float64 {
 	return 0
 }
 
@@ -262,3 +273,20 @@ func TestLightAvailability(bServ blockservice.BlockService) *LightAvailability {
 func TestFullAvailability(bServ blockservice.BlockService) *FullAvailability {
 	return NewFullAvailability(bServ, routing.NewRoutingDiscovery(routinghelpers.Null{}), nil)
 }
+
+type TestSuccessfulAvailability struct {
+}
+
+// NewTestSuccessfulAvailability returns an Availability that always
+// returns successfully when SharesAvailable is called.
+func NewTestSuccessfulAvailability() Availability {
+	return &TestSuccessfulAvailability{}
+}
+
+func (tsa *TestSuccessfulAvailability) SharesAvailable(context.Context, *Root) error {
+	return nil
+}
+
+func (tsa *TestSuccessfulAvailability) ProbabilityOfAvailability() float64 {
+	return 0
+}