Repository: ObolNetwork/charon

Issue:
Secure and Simplify Import of .charon Folders for DappNode Integration
## 🎯 Problem to be solved

Currently, there is not an easy way for someone to load charon artifacts into the dappnode package. 

Explore methods for importing a pre-created `.charon` folder into a DappNode. DappNode operates through a web UI, which makes it difficult to upload folders/directories directly. The current workaround involves zipping the `.charon` folder, uploading it, and then unzipping it on the server side. 

## 🛠️ Proposed solution
- [x] Consider implementing an optional flag (e.g., `charon create cluster --zipped`) to automatically output node subfolders as `.tar.xz` files
- [x] ~~Confirm with the DappNode team the feasibility of uploading folders/directories through the web UI.~~ Confirmed that they do not support folder upload, and only support file upload. 
- [x] ~~Investigate the support of using SSH to transfer the `.charon` folder despite the general non-reliance on SSH for DappNode UI access.~~ Not a preferred approach
- [x] ~~Address security concerns by exploring the addition of export/import commands that zip and encrypt the data~~ What we will do is match our output to their backup output. (which is a `.tar.xz` of the `.charon` folder)

## To test it

The existing dappnode package will unzip and install a backup via its restore process. Currently it uses the file name to decide which cluster slot to load this into. A feature we add to `.charon` to export as `.tar.xz` could be tested by renaming the output to what the package wants to see and "restoring" it to an existing package, overwriting the enr that may have been there unused. 

(In the modified version, they will have a button per cluster slot to upload specifically to, negating the need for a precise file name. )

Discussion:

Solution patch:
diff --git a/.github/workflows/golangci-lint.yml b/.github/workflows/golangci-lint.yml
index 5a58dd41..5a7e4d81 100644
--- a/.github/workflows/golangci-lint.yml
+++ b/.github/workflows/golangci-lint.yml
@@ -17,9 +17,9 @@ jobs:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/setup-go
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v7
+        uses: golangci/golangci-lint-action@v8
         with:
-          version: v2.0.1
+          version: v2.1.6
       - name: notify failure
         if: failure() && github.ref == 'refs/heads/main'
         env:
diff --git a/.golangci.yml b/.golangci.yml
index c4ce2bff..169d0a52 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -11,6 +11,7 @@ linters:
     - exhaustruct
     - forcetypeassert
     - funlen
+    - funcorder
     - gochecknoglobals
     - gocognit
     - gocyclo
diff --git a/.pre-commit/run_linter.sh b/.pre-commit/run_linter.sh
index ad651328..8b57665e 100755
--- a/.pre-commit/run_linter.sh
+++ b/.pre-commit/run_linter.sh
@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-VERSION="2.0.1"
+VERSION="2.1.6"
 
 if ! command -v golangci-lint &>/dev/null; then
     echo "golangci-lint could not be found"
diff --git a/README.md b/README.md
index bb96bfd3..c5a13090 100644
--- a/README.md
+++ b/README.md
@@ -25,7 +25,7 @@ which contains a docker compose setup for running a full charon cluster on your
 ## Documentation
 
 The [Obol Docs](https://docs.obol.tech/) website is the best place to get started.
-The important sections are [intro](https://docs.obol.tech/docs/intro),
+The important sections are [intro](https://docs.obol.org/learn/charon),
 [key concepts](https://docs.obol.tech/docs/int/key-concepts) and [charon](https://docs.obol.tech/docs/charon/intro).
 
 For detailed documentation on this repo, see the [docs](docs) folder:
diff --git a/app/app.go b/app/app.go
index 55a78419..f482f9c3 100644
--- a/app/app.go
+++ b/app/app.go
@@ -455,7 +455,7 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 	if err != nil {
 		return err
 	}
-	sseListener.SubscribeChainReorgEvent(sched.ChainReorgEventHandler)
+	sseListener.SubscribeChainReorgEvent(sched.HandleChainReorgEvent)
 
 	feeRecipientFunc := func(pubkey core.PubKey) string {
 		return feeRecipientAddrByCorePubkey[pubkey]
diff --git a/app/featureset/featureset.go b/app/featureset/featureset.go
index af4e60e1..95eba80d 100644
--- a/app/featureset/featureset.go
+++ b/app/featureset/featureset.go
@@ -49,6 +49,9 @@ const (
 	// Linear enables Linear round timer for consensus rounds.
 	// When active has precedence over EagerDoubleLinear round timer.
 	Linear Feature = "linear"
+
+	// SSEReorgDuties enables Scheduler to refresh duties when reorg occurs.
+	SSEReorgDuties Feature = "sse_reorg_duties"
 )
 
 var (
@@ -61,6 +64,7 @@ var (
 		JSONRequests:         statusAlpha,
 		GnosisBlockHotfix:    statusAlpha,
 		Linear:               statusAlpha,
+		SSEReorgDuties:       statusAlpha,
 		// Add all features and there status here.
 	}
 
diff --git a/app/log/config.go b/app/log/config.go
index 3109a104..e01f3192 100644
--- a/app/log/config.go
+++ b/app/log/config.go
@@ -177,7 +177,7 @@ func InitLogger(config Config) error {
 	})
 
 	if registerError != nil {
-		return errors.Wrap(err, "zap register sink lumberjack")
+		return errors.Wrap(registerError, "zap register sink lumberjack")
 	}
 
 	writer, _, err := zap.Open("stderr")
diff --git a/app/sse/listener.go b/app/sse/listener.go
index 865e45b6..6469c47d 100644
--- a/app/sse/listener.go
+++ b/app/sse/listener.go
@@ -125,6 +125,13 @@ func (p *listener) handleHeadEvent(ctx context.Context, event *event, addr strin
 
 	sseHeadSlotGauge.WithLabelValues(addr).Set(float64(slot))
 
+	log.Debug(ctx, "SSE head event",
+		z.U64("slot", slot),
+		z.Str("delay", delay.String()),
+		z.Str("block", head.Block),
+		z.Str("prev_ddr", head.PreviousDutyDependentRoot),
+		z.Str("curr_ddr", head.CurrentDutyDependentRoot))
+
 	return nil
 }
 
@@ -138,9 +145,6 @@ func (p *listener) handleChainReorgEvent(ctx context.Context, event *event, addr
 	if err != nil {
 		return errors.Wrap(err, "parse slot to uint64", z.Str("addr", addr))
 	}
-	if slot > math.MaxInt64 {
-		return errors.New("slot value exceeds int64 range", z.Str("addr", addr), z.U64("slot", slot))
-	}
 	depth, err := strconv.ParseUint(chainReorg.Depth, 10, 64)
 	if err != nil {
 		return errors.Wrap(err, "parse depth to uint64", z.Str("addr", addr))
@@ -153,7 +157,13 @@ func (p *listener) handleChainReorgEvent(ctx context.Context, event *event, addr
 	reorgEpoch := (slot - depth) / p.slotsPerEpoch
 	p.notifyChainReorg(ctx, eth2p0.Epoch(reorgEpoch))
 
-	log.Debug(ctx, "Beacon node reorged", z.U64("slot", slot), z.U64("depth", depth))
+	log.Debug(ctx, "SSE chain reorg event",
+		z.U64("slot", slot),
+		z.Str("epoch", chainReorg.Epoch),
+		z.U64("reorg_epoch", reorgEpoch),
+		z.U64("depth", depth),
+		z.Str("old_head_block", chainReorg.OldHeadBlock),
+		z.Str("new_head_block", chainReorg.NewHeadBlock))
 
 	sseChainReorgDepthGauge.WithLabelValues(addr).Set(float64(depth))
 
diff --git a/cmd/createcluster.go b/cmd/createcluster.go
index 8cd7eabb..d7478233 100644
--- a/cmd/createcluster.go
+++ b/cmd/createcluster.go
@@ -3,7 +3,9 @@
 package cmd
 
 import (
+	"archive/tar"
 	"bytes"
+	"compress/gzip"
 	"context"
 	"crypto/rand"
 	"encoding/hex"
@@ -91,6 +93,8 @@ type clusterConfig struct {
 	testnetConfig eth2util.Network
 
 	ExecutionEngineAddr string
+
+	Zipped bool
 }
 
 func newCreateClusterCmd(runFunc func(context.Context, io.Writer, clusterConfig) error) *cobra.Command {
@@ -153,6 +157,7 @@ func bindClusterFlags(flags *pflag.FlagSet, config *clusterConfig) {
 	flags.UintVar(&config.TargetGasLimit, "target-gas-limit", 36000000, "Preferred target gas limit for transactions.")
 	flags.BoolVar(&config.Compounding, "compounding", false, "Enable compounding rewards for validators by using 0x02 withdrawal credentials.")
 	flags.StringVar(&config.ExecutionEngineAddr, "execution-client-rpc-endpoint", "", "The address of the execution engine JSON-RPC API.")
+	flags.BoolVar(&config.Zipped, "zipped", false, "Create a tar archive compressed with gzip of the cluster directory after creation.")
 }
 
 func bindInsecureFlags(flags *pflag.FlagSet, insecureKeys *bool) {
@@ -338,11 +343,17 @@ func runCreateCluster(ctx context.Context, w io.Writer, conf clusterConfig) erro
 		return err
 	}
 
+	if conf.Zipped {
+		if err = bundleOutput(conf.ClusterDir, numNodes); err != nil {
+			return err
+		}
+	}
+
 	if conf.SplitKeys {
 		writeWarning(w)
 	}
 
-	if err := writeOutput(w, conf.SplitKeys, conf.ClusterDir, numNodes, keysToDisk); err != nil {
+	if err := writeOutput(w, conf.SplitKeys, conf.ClusterDir, numNodes, keysToDisk, conf.Zipped); err != nil {
 		return err
 	}
 
@@ -901,7 +912,7 @@ func newPeer(clusterDir string, peerIdx int) (enr.Record, *k1.PrivateKey, error)
 }
 
 // writeOutput writes the cluster generation output.
-func writeOutput(out io.Writer, splitKeys bool, clusterDir string, numNodes int, keysToDisk bool) error {
+func writeOutput(out io.Writer, splitKeys bool, clusterDir string, numNodes int, keysToDisk, zipped bool) error {
 	absClusterDir, err := filepath.Abs(clusterDir)
 	if err != nil {
 		return errors.Wrap(err, "absolute path retrieval")
@@ -921,6 +932,9 @@ func writeOutput(out io.Writer, splitKeys bool, clusterDir string, numNodes int,
 		_, _ = sb.WriteString("│  │  ├─ keystore-*.json\tValidator private share key for duty signing\n")
 		_, _ = sb.WriteString("│  │  ├─ keystore-*.txt\t\tKeystore password files for keystore-*.json\n")
 	}
+	if zipped {
+		_, _ = sb.WriteString(fmt.Sprintf("\nFiles compressed and archived to:\n%s/cluster.tar.gz\n", absClusterDir))
+	}
 
 	_, _ = fmt.Fprint(out, sb.String())
 
@@ -1178,3 +1192,70 @@ func validateNetworkConfig(conf clusterConfig) error {
 
 	return errors.New("missing --network flag or testnet config flags")
 }
+
+// bundleOutput archives all node directories (node0, node1, ..., nodeN) within targetDir into a gzipped tarball named "cluster.tar.gz" in targetDir.
+// After successfully creating the archive, it deletes the original node directories from disk.
+func bundleOutput(targetDir string, numNodes int) error {
+	file, err := os.Create(filepath.Join(targetDir, "cluster.tar.gz"))
+	if err != nil {
+		return errors.Wrap(err, "create .tar.gz file")
+	}
+	defer file.Close()
+
+	gzw := gzip.NewWriter(file)
+	defer gzw.Close()
+
+	tw := tar.NewWriter(gzw)
+	defer tw.Close()
+
+	dirs := []string{}
+	for i := range numNodes {
+		dirs = append(dirs, filepath.Join(targetDir, fmt.Sprintf("node%d", i)))
+	}
+
+	for _, dir := range dirs {
+		err = filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+			if err != nil {
+				return errors.Wrap(err, "filepath walk")
+			}
+			if !info.Mode().IsRegular() {
+				return nil
+			}
+			relPath, err := filepath.Rel(targetDir, path)
+			if err != nil {
+				return errors.Wrap(err, "relative path")
+			}
+			header, err := tar.FileInfoHeader(info, info.Name())
+			if err != nil {
+				return errors.Wrap(err, "file info header")
+			}
+			header.Name = relPath
+			if err := tw.WriteHeader(header); err != nil {
+				return errors.Wrap(err, "write header")
+			}
+			f, err := os.Open(path)
+			if err != nil {
+				return errors.Wrap(err, "open file")
+			}
+			defer f.Close()
+			_, err = io.Copy(tw, f)
+			if err != nil {
+				return errors.Wrap(err, "copy file", z.Str("filename", path))
+			}
+
+			return nil
+		})
+		if err != nil {
+			return errors.Wrap(err, "filepath walk")
+		}
+	}
+
+	for _, dir := range dirs {
+		err := os.RemoveAll(dir)
+		if err != nil {
+			return errors.Wrap(err, "remove file")
+		}
+	}
+
+	return nil
+}
diff --git a/core/scheduler/scheduler.go b/core/scheduler/scheduler.go
index 66d835af..a88bb383 100644
--- a/core/scheduler/scheduler.go
+++ b/core/scheduler/scheduler.go
@@ -19,6 +19,7 @@ import (
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/eth2wrap"
 	"github.com/obolnetwork/charon/app/expbackoff"
+	"github.com/obolnetwork/charon/app/featureset"
 	"github.com/obolnetwork/charon/app/log"
 	"github.com/obolnetwork/charon/app/z"
 	"github.com/obolnetwork/charon/core"
@@ -29,17 +30,22 @@ const trimEpochOffset = 3 // Trim cached duties after 3 epochs. Note inclusion d
 // delayFunc abstracts slot offset delaying/sleeping for deterministic tests.
 type delayFunc func(duty core.Duty, deadline time.Time) <-chan time.Time
 
+// schedSlotFunc is a function type that is called before every scheduled slot.
+// Used only in testing.
+type schedSlotFunc func(ctx context.Context, slot core.Slot)
+
 // NewForT returns a new scheduler for testing using a fake clock.
 func NewForT(t *testing.T, clock clockwork.Clock, delayFunc delayFunc, pubkeys []core.PubKey,
-	eth2Cl eth2wrap.Client, builderAPI bool,
+	eth2Cl eth2wrap.Client, schedSlotFunc schedSlotFunc,
 ) *Scheduler {
 	t.Helper()
 
-	s, err := New(pubkeys, eth2Cl, builderAPI)
+	s, err := New(pubkeys, eth2Cl, false)
 	require.NoError(t, err)
 
 	s.clock = clock
 	s.delayFunc = delayFunc
+	s.schedSlotFunc = schedSlotFunc
 
 	return s
 }
@@ -58,6 +64,7 @@ func New(pubkeys []core.PubKey, eth2Cl eth2wrap.Client, builderEnabled bool) (*S
 		},
 		metricSubmitter: newMetricSubmitter(),
 		resolvedEpoch:   math.MaxInt64,
+		resolvingEpoch:  math.MaxInt64,
 		builderEnabled:  builderEnabled,
 	}, nil
 }
@@ -70,12 +77,14 @@ type Scheduler struct {
 	delayFunc       delayFunc
 	metricSubmitter metricSubmitter
 	resolvedEpoch   uint64
+	resolvingEpoch  uint64
 	duties          map[core.Duty]core.DutyDefinitionSet
 	dutiesByEpoch   map[uint64][]core.Duty
 	dutiesMutex     sync.RWMutex
 	dutySubs        []func(context.Context, core.Duty, core.DutyDefinitionSet) error
 	slotSubs        []func(context.Context, core.Slot) error
 	builderEnabled  bool
+	schedSlotFunc   schedSlotFunc
 }
 
 // SubscribeDuties subscribes a callback function for triggered duties.
@@ -127,10 +136,21 @@ func (s *Scheduler) Run() error {
 	}
 }
 
-// ChainReorgEventHandler is connected to SSE Listener and handles chain reorg events.
-func (*Scheduler) ChainReorgEventHandler(ctx context.Context, epoch eth2p0.Epoch) {
-	// TODO: implement chain reorg handling.
-	log.Warn(ctx, "Scheduler received chain reorg event", nil, z.U64("epoch", uint64(epoch)))
+// HandleChainReorgEvent is connected to SSE Listener and handles chain reorg events.
+func (s *Scheduler) HandleChainReorgEvent(ctx context.Context, epoch eth2p0.Epoch) {
+	if featureset.Enabled(featureset.SSEReorgDuties) {
+		resolvedEpoch := s.getResolvedEpoch()
+		if uint64(epoch) < resolvedEpoch {
+			// Removing current epoch duties, because of a chain reorg.
+			s.trimDuties(resolvedEpoch)
+			// Duties are to be resolved again in the next slot by scheduleSlot().
+			s.setResolvedEpoch(math.MaxInt64)
+
+			log.Info(ctx, "Chain reorg event handled, duties trimmed", z.U64("reorg_epoch", uint64(epoch)), z.U64("resolved_epoch", resolvedEpoch))
+		}
+	} else {
+		log.Warn(ctx, "Chain reorg event ignored due to disabled SSEReorgDuties feature", nil, z.U64("reorg_epoch", uint64(epoch)))
+	}
 }
 
 // emitCoreSlot calls all slot subscriptions asynchronously with the provided slot.
@@ -158,6 +178,16 @@ func (s *Scheduler) GetDutyDefinition(ctx context.Context, duty core.Duty) (core
 	}
 
 	epoch := duty.Slot / slotsPerEpoch
+
+	// This has to be very rare event, when the requested epoch is being resolved.
+	// We wait for the epoch to be resolved before returning the duty definition.
+	for s.isResolvingEpoch(epoch) {
+		if ctx.Err() != nil {
+			return nil, errors.Wrap(ctx.Err(), "context cancelled while waiting for epoch to resolve")
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+
 	if !s.isEpochResolved(epoch) {
 		return nil, errors.New("epoch not resolved yet",
 			z.Str("duty", duty.String()), z.U64("epoch", epoch))
@@ -178,7 +208,13 @@ func (s *Scheduler) GetDutyDefinition(ctx context.Context, duty core.Duty) (core
 
 // scheduleSlot resolves upcoming duties and triggers resolved duties for the slot.
 func (s *Scheduler) scheduleSlot(ctx context.Context, slot core.Slot) {
+	if s.schedSlotFunc != nil {
+		s.schedSlotFunc(ctx, slot)
+	}
+
 	if s.getResolvedEpoch() != slot.Epoch() {
+		log.Debug(ctx, "Resolving duties for slot", z.U64("slot", slot.Slot), z.U64("epoch", slot.Epoch()))
+
 		err := s.resolveDuties(ctx, slot)
 		if err != nil {
 			log.Warn(ctx, "Resolving duties error (retrying next slot)", err, z.U64("slot", slot.Slot))
@@ -255,6 +291,9 @@ func delaySlotOffset(ctx context.Context, slot core.Slot, duty core.Duty, delayF
 
 // resolveDuties resolves the duties for the slot's epoch, caching the results.
 func (s *Scheduler) resolveDuties(ctx context.Context, slot core.Slot) error {
+	s.setResolvingEpoch(slot.Epoch())
+	defer s.setResolvingEpoch(math.MaxInt64)
+
 	vals, err := resolveActiveValidators(ctx, s.eth2Cl, s.metricSubmitter, slot.Epoch())
 	if err != nil {
 		return err
@@ -519,6 +558,24 @@ func (s *Scheduler) setResolvedEpoch(epoch uint64) {
 	s.resolvedEpoch = epoch
 }
 
+func (s *Scheduler) setResolvingEpoch(epoch uint64) {
+	s.dutiesMutex.Lock()
+	defer s.dutiesMutex.Unlock()
+
+	s.resolvingEpoch = epoch
+}
+
+func (s *Scheduler) isResolvingEpoch(epoch uint64) bool {
+	s.dutiesMutex.RLock()
+	defer s.dutiesMutex.RUnlock()
+
+	if s.resolvingEpoch == math.MaxInt64 {
+		return false
+	}
+
+	return s.resolvingEpoch == epoch
+}
+
 // isEpochResolved returns true if the epoch is resolved.
 func (s *Scheduler) isEpochResolved(epoch uint64) bool {
 	if s.getResolvedEpoch() == math.MaxInt64 {
@@ -566,6 +623,10 @@ func newSlotTicker(ctx context.Context, eth2Cl eth2wrap.Client, clock clockwork.
 		return nil, err
 	}
 
+	if slotDuration == 0 {
+		return nil, errors.New("slot duration cannot be zero")
+	}
+
 	currentSlot := func() core.Slot {
 		chainAge := clock.Since(genesisTime)
 		slot := int64(chainAge / slotDuration)



Test patch:
diff --git a/app/featureset/featureset_internal_test.go b/app/featureset/featureset_internal_test.go
index cee78730..424cd280 100644
--- a/app/featureset/featureset_internal_test.go
+++ b/app/featureset/featureset_internal_test.go
@@ -17,6 +17,7 @@ func TestAllFeatureStatus(t *testing.T) {
 		JSONRequests,
 		GnosisBlockHotfix,
 		Linear,
+		SSEReorgDuties,
 	}
 
 	for _, feature := range features {
diff --git a/app/sse/listener_internal_test.go b/app/sse/listener_internal_test.go
index a0b7fff0..43d89e44 100644
--- a/app/sse/listener_internal_test.go
+++ b/app/sse/listener_internal_test.go
@@ -90,8 +90,8 @@ func TestHandleEvents(t *testing.T) {
 			l := &listener{
 				chainReorgSubs: make([]ChainReorgEventHandlerFunc, 0),
 				slotDuration:   12 * time.Second,
-				genesisTime:    time.Date(2020, 12, 1, 12, 0, 23, 0, time.UTC),
 				slotsPerEpoch:  32,
+				genesisTime:    time.Date(2020, 12, 1, 12, 0, 23, 0, time.UTC),
 			}
 
 			err := l.eventHandler(t.Context(), test.event, "test")
diff --git a/cmd/createcluster_internal_test.go b/cmd/createcluster_internal_test.go
index ebde67b7..e8f6b1ec 100644
--- a/cmd/createcluster_internal_test.go
+++ b/cmd/createcluster_internal_test.go
@@ -3,7 +3,9 @@
 package cmd
 
 import (
+	"archive/tar"
 	"bytes"
+	"compress/gzip"
 	"context"
 	"encoding/hex"
 	"encoding/json"
@@ -956,6 +958,94 @@ func TestClusterCLI(t *testing.T) {
 	}
 }
 
+func TestZipping(t *testing.T) {
+	ctx := t.Context()
+	conf := clusterConfig{
+		Name:              "test",
+		NumNodes:          4,
+		NumDVs:            4,
+		Threshold:         3,
+		TargetGasLimit:    30000000,
+		Network:           eth2util.Goerli.Name,
+		WithdrawalAddrs:   []string{zeroAddress},
+		FeeRecipientAddrs: []string{zeroAddress},
+		InsecureKeys:      true,
+	}
+
+	conf.ClusterDir = t.TempDir()
+
+	var buf bytes.Buffer
+	require.NoError(t, runCreateCluster(ctx, &buf, conf))
+
+	require.NoError(t, bundleOutput(conf.ClusterDir, conf.NumNodes))
+
+	unzippedDir := t.TempDir()
+	require.NoError(t, unzipOutputT(t, conf.ClusterDir, unzippedDir))
+
+	err := os.RemoveAll(filepath.Join(conf.ClusterDir, "cluster.tar.gz"))
+	require.NoError(t, err)
+
+	// Walk both directories and compare files
+	err = filepath.Walk(conf.ClusterDir, func(path string, info os.FileInfo, err error) error {
+		require.NoError(t, err)
+
+		// Get the corresponding file in the unzipped directory
+		relPath, err := filepath.Rel(conf.ClusterDir, path)
+		require.NoError(t, err, "failed to get relative path")
+		unzippedPath := filepath.Join(unzippedDir, relPath)
+
+		if info.IsDir() {
+			require.NoError(t, os.MkdirAll(unzippedPath, info.Mode()))
+			return nil
+		}
+
+		unzippedInfo, err := os.Stat(unzippedPath)
+		require.NoError(t, err, "failed to stat unzipped file")
+
+		require.Equal(t, info.Size(), unzippedInfo.Size(), "mismatched files: %s and %s", path, unzippedPath)
+
+		return nil
+	})
+	require.NoError(t, err, "unzipped directory should match original directory structure")
+}
+
+func unzipOutputT(t *testing.T, sourceDir, targetDir string) error {
+	t.Helper()
+
+	zf, err := os.Open(filepath.Join(sourceDir, "cluster.tar.gz"))
+	require.NoError(t, err, "failed to open archive")
+	defer zf.Close()
+
+	gzr, err := gzip.NewReader(zf)
+	require.NoError(t, err, "failed to create gzip reader")
+	defer gzr.Close()
+
+	tr := tar.NewReader(gzr)
+
+	for {
+		header, err := tr.Next()
+		if err == io.EOF {
+			return nil // End of archive
+		}
+		require.NoError(t, err, "tar read error")
+
+		target := filepath.Join(targetDir, header.Name)
+		switch header.Typeflag {
+		case tar.TypeDir:
+			require.NoError(t, os.MkdirAll(target, header.FileInfo().Mode()), "failed to create directory")
+		case tar.TypeReg:
+			require.NoError(t, os.MkdirAll(filepath.Dir(target), 0o755), "failed to create parent directory")
+			f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY, header.FileInfo().Mode())
+			require.NoError(t, err, "failed to create file")
+			_, err = io.Copy(f, tr)
+			require.NoError(t, err, "failed to copy file contents")
+			require.NoError(t, f.Close(), "failed to close file")
+		default:
+			// Ignore other types
+		}
+	}
+}
+
 // mockKeymanagerReq is a mock keymanager request for use in tests.
 type mockKeymanagerReq struct {
 	Keystores []string `json:"keystores"`
diff --git a/cmd/testinfra.go b/cmd/testinfra.go
index 71591712..8293fa1e 100644
--- a/cmd/testinfra.go
+++ b/cmd/testinfra.go
@@ -683,7 +683,7 @@ func availableMemoryLinux(context.Context) (int64, error) {
 	}
 	scanner := bufio.NewScanner(file)
 	if scanner.Err() != nil {
-		return 0, errors.Wrap(err, "new scanner")
+		return 0, errors.Wrap(scanner.Err(), "new scanner")
 	}
 
 	for scanner.Scan() {
@@ -721,7 +721,7 @@ func availableMemoryMacos(ctx context.Context) (int64, error) {
 	outBuf := bytes.NewBuffer(out)
 	scanner := bufio.NewScanner(outBuf)
 	if scanner.Err() != nil {
-		return 0, errors.Wrap(err, "new scanner")
+		return 0, errors.Wrap(scanner.Err(), "new scanner")
 	}
 
 	var pagesFree, pagesInactive, pagesSpeculative int64
@@ -763,7 +763,7 @@ func totalMemoryLinux(context.Context) (int64, error) {
 	}
 	scanner := bufio.NewScanner(file)
 	if scanner.Err() != nil {
-		return 0, errors.Wrap(err, "new scanner")
+		return 0, errors.Wrap(scanner.Err(), "new scanner")
 	}
 
 	for scanner.Scan() {
diff --git a/core/scheduler/scheduler_internal_test.go b/core/scheduler/scheduler_internal_test.go
index 2d03c18b..496a006c 100644
--- a/core/scheduler/scheduler_internal_test.go
+++ b/core/scheduler/scheduler_internal_test.go
@@ -127,3 +127,15 @@ func TestResolveSyncCommDuties(t *testing.T) {
 		SlotsPerEpoch: 1,
 	}, schedVals), "invalid sync committee duty pubkey")
 }
+
+func TestResolvingEpoch(t *testing.T) {
+	sched, _ := setupScheduler(t)
+
+	sched.setResolvingEpoch(10)
+	require.True(t, sched.isResolvingEpoch(10))
+	require.False(t, sched.isResolvingEpoch(11))
+
+	sched.setResolvingEpoch(11)
+	require.False(t, sched.isResolvingEpoch(10))
+	require.True(t, sched.isResolvingEpoch(11))
+}
diff --git a/core/scheduler/scheduler_test.go b/core/scheduler/scheduler_test.go
index 086a4945..9320d9a6 100644
--- a/core/scheduler/scheduler_test.go
+++ b/core/scheduler/scheduler_test.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/eth2wrap"
+	"github.com/obolnetwork/charon/app/featureset"
 	"github.com/obolnetwork/charon/core"
 	"github.com/obolnetwork/charon/core/scheduler"
 	"github.com/obolnetwork/charon/testutil"
@@ -107,7 +108,7 @@ func TestSchedulerWait(t *testing.T) {
 		{
 			Name:       "synced errors",
 			SyncedErrs: 10,
-			WaitSecs:   15, // We use expbackoff.FastConfig
+			WaitSecs:   14, // We use expbackoff.FastConfig
 		},
 	}
 
@@ -145,7 +146,7 @@ func TestSchedulerWait(t *testing.T) {
 			}
 
 			dd := new(delayer)
-			sched := scheduler.NewForT(t, clock, dd.delay, nil, eth2Cl, false)
+			sched := scheduler.NewForT(t, clock, dd.delay, nil, eth2Cl, nil)
 			sched.Stop() // Just run wait functions, then quit.
 			require.NoError(t, sched.Run())
 			require.LessOrEqual(t, test.WaitSecs, int(clock.Since(t0).Seconds()))
@@ -216,11 +217,11 @@ func TestSchedulerDuties(t *testing.T) {
 			// Construct scheduler
 			clock := newTestClock(t0)
 			delayer := new(delayer)
-			sched := scheduler.NewForT(t, clock, delayer.delay, pubkeys, eth2Cl, false)
+			sched := scheduler.NewForT(t, clock, delayer.delay, pubkeys, eth2Cl, nil)
 
 			// Only test scheduler output for first N slots, so Stop scheduler (and slotTicker) after that.
 			const stopAfter = 3
-			eth2Resp, err := eth2Cl.Spec(context.Background(), &eth2api.SpecOpts{})
+			eth2Resp, err := eth2Cl.Spec(t.Context(), &eth2api.SpecOpts{})
 			require.NoError(t, err)
 
 			slotDuration, ok := eth2Resp.Data["SECONDS_PER_SLOT"].(time.Duration)
@@ -291,7 +292,7 @@ func TestSchedulerDuties(t *testing.T) {
 
 func TestScheduler_GetDuty(t *testing.T) {
 	var (
-		ctx    = context.Background()
+		ctx    = t.Context()
 		t0     time.Time
 		slot   = uint64(1)
 		valSet = beaconmock.ValidatorSetA
@@ -314,7 +315,7 @@ func TestScheduler_GetDuty(t *testing.T) {
 	// Construct scheduler.
 	clock := newTestClock(t0)
 	dd := new(delayer)
-	sched := scheduler.NewForT(t, clock, dd.delay, pubkeys, eth2Cl, false)
+	sched := scheduler.NewForT(t, clock, dd.delay, pubkeys, eth2Cl, nil)
 
 	_, err = sched.GetDutyDefinition(ctx, core.NewAttesterDuty(slot))
 	require.ErrorContains(t, err, "epoch not resolved yet")
@@ -331,12 +332,9 @@ func TestScheduler_GetDuty(t *testing.T) {
 	})
 	require.ErrorIs(t, err, core.ErrDeprecatedDutyBuilderProposer)
 
-	eth2Resp, err := eth2Cl.Spec(ctx, &eth2api.SpecOpts{})
+	slotDuration, slotsPerEpoch, err := eth2wrap.FetchSlotsConfig(ctx, eth2Cl)
 	require.NoError(t, err)
 
-	slotDuration, ok := eth2Resp.Data["SECONDS_PER_SLOT"].(time.Duration)
-	require.True(t, ok)
-
 	clock.CallbackAfter(t0.Add(slotDuration).Add(time.Second), func() {
 		res, err := sched.GetDutyDefinition(ctx, core.NewAttesterDuty(slot))
 		require.NoError(t, err)
@@ -361,9 +359,6 @@ func TestScheduler_GetDuty(t *testing.T) {
 		}
 	})
 
-	slotsPerEpoch, err := eth2Cl.SlotsPerEpoch(ctx)
-	require.NoError(t, err)
-
 	// Expire all duties
 	const trimEpochOffset = 3
 	expiry := time.Duration(trimEpochOffset*slotsPerEpoch) * slotDuration
@@ -388,7 +383,7 @@ func TestScheduler_GetDuty(t *testing.T) {
 
 func TestNoActive(t *testing.T) {
 	var (
-		ctx          = context.Background()
+		ctx          = t.Context()
 		t0           = time.Now()
 		slotDuration = time.Second
 	)
@@ -404,7 +399,7 @@ func TestNoActive(t *testing.T) {
 	// Construct scheduler.
 	clock := newTestClock(t0)
 	dd := new(delayer)
-	sched := scheduler.NewForT(t, clock, dd.delay, nil, eth2Cl, false)
+	sched := scheduler.NewForT(t, clock, dd.delay, nil, eth2Cl, nil)
 
 	clock.CallbackAfter(t0.Add(slotDuration*2), func() {
 		_, err := sched.GetDutyDefinition(ctx, core.NewAttesterDuty(1))
@@ -415,6 +410,65 @@ func TestNoActive(t *testing.T) {
 	require.NoError(t, sched.Run())
 }
 
+func TestHandleChainReorgEvent(t *testing.T) {
+	var (
+		t0     time.Time
+		valSet = beaconmock.ValidatorSetA
+	)
+
+	featureset.EnableForT(t, featureset.SSEReorgDuties)
+
+	// Configure beacon mock.
+	eth2Cl, err := beaconmock.New(
+		beaconmock.WithValidatorSet(valSet),
+		beaconmock.WithGenesisTime(t0),
+		beaconmock.WithDeterministicAttesterDuties(1),
+		beaconmock.WithSlotsPerEpoch(4),
+	)
+	require.NoError(t, err)
+
+	// Get pubkeys for validators to schedule.
+	pubkeys, err := valSet.CorePubKeys()
+	require.NoError(t, err)
+
+	// Construct scheduler.
+	schedSlotCh := make(chan core.Slot)
+	schedSlotFunc := func(ctx context.Context, slot core.Slot) {
+		select {
+		case <-ctx.Done():
+			return
+		case schedSlotCh <- slot:
+		}
+	}
+	clock := newTestClock(t0)
+	dd := new(delayer)
+	sched := scheduler.NewForT(t, clock, dd.delay, pubkeys, eth2Cl, schedSlotFunc)
+
+	doneCh := make(chan error, 1)
+	go func() {
+		doneCh <- sched.Run()
+		close(schedSlotCh)
+	}()
+
+	for slot := range schedSlotCh {
+		switch slot.Slot {
+		case 1: // epoch 0
+			_, err := sched.GetDutyDefinition(t.Context(), core.NewAttesterDuty(1))
+			require.NoError(t, err)
+		case 5: // epoch 1
+			sched.HandleChainReorgEvent(t.Context(), 0)
+			_, err = sched.GetDutyDefinition(t.Context(), core.NewAttesterDuty(5))
+			require.ErrorContains(t, err, "epoch not resolved yet")
+		case 7: // epoch 1 after reorg
+			_, err := sched.GetDutyDefinition(t.Context(), core.NewAttesterDuty(6))
+			require.NoError(t, err)
+			sched.Stop()
+		}
+	}
+
+	require.NoError(t, <-doneCh)
+}
+
 // delayer implements scheduler.delayFunc and records the deadline and returns it immediately.
 type delayer struct {
 	mu        sync.Mutex