Remove explicit pause/resume from the manifest sender

manifest/dynamic_manifest.go

@@ -42,7 +42,7 @@ type ManifestUpdateMessage struct {
 	// An increasing sequence number for ordering manifest updates received over the network.
 	MessageSequence uint64
 	// The manifest to apply or nil to pause the network.
-	Manifest *Manifest
+	Manifest Manifest
 }
 
 func (m ManifestUpdateMessage) Marshal() ([]byte, error) {
@@ -124,7 +124,7 @@ func (m *DynamicManifestProvider) Start(startCtx context.Context) error {
 		}
 
 		msgSeqNumber = update.MessageSequence
-		currentManifest = update.Manifest
+		currentManifest = &update.Manifest
 	}
 
 	m.manifestChanges <- currentManifest
@@ -173,7 +173,8 @@ func (m *DynamicManifestProvider) Start(startCtx context.Context) error {
 			msgSeqNumber = update.MessageSequence
 
 			oldManifest := currentManifest
-			currentManifest = update.Manifest
+			manifestCopy := update.Manifest
+			currentManifest = &manifestCopy
 
 			// If we're receiving the same manifest multiple times (manifest publisher
 			// could have restarted), don't re-apply it.
@@ -182,7 +183,7 @@ func (m *DynamicManifestProvider) Start(startCtx context.Context) error {
 			}
 
 			select {
-			case m.manifestChanges <- update.Manifest:
+			case m.manifestChanges <- currentManifest:
 			case <-m.runningCtx.Done():
 				return nil
 			}

manifest/dynamic_manifest_test.go

@@ -68,12 +68,14 @@ func TestDynamicManifest(t *testing.T) {
 	// Should receive the initial manifest.
 	require.True(t, initialManifest.Equal(<-provider.ManifestUpdates()))
 
-	// Pausing should send nil.
-	sender.Pause()
-	require.Nil(t, <-provider.ManifestUpdates())
+	// Pausing should send an update.
+	pausedManifest := *initialManifest
+	pausedManifest.Pause = true
+	sender.UpdateManifest(&pausedManifest)
+	require.True(t, pausedManifest.Equal(<-provider.ManifestUpdates()))
 
 	// Should get the initial manifest again.
-	sender.Resume()
+	sender.UpdateManifest(initialManifest)
 	require.True(t, initialManifest.Equal(<-provider.ManifestUpdates()))
 
 	cancelSender()

manifest/manifest.go

@@ -131,6 +131,7 @@ func (m *Manifest) Equal(o *Manifest) bool {
 	}
 
 	return m.NetworkName == o.NetworkName &&
+		m.Pause == o.Pause &&
 		m.InitialInstance == o.InitialInstance &&
 		m.BootstrapEpoch == o.BootstrapEpoch &&
 		m.IgnoreECPower == o.IgnoreECPower &&

manifest/manifest_sender.go

@@ -25,7 +25,6 @@ type ManifestSender struct {
 	lk       sync.Mutex
 	manifest *Manifest
 	msgSeq   uint64
-	paused   bool
 }
 
 func NewManifestSender(ctx context.Context, h host.Host, ps *pubsub.PubSub, firstManifest *Manifest, pubishInterval time.Duration) (*ManifestSender, error) {
@@ -81,11 +80,14 @@ func (m *ManifestSender) publishManifest(ctx context.Context) error {
 	m.lk.Lock()
 	defer m.lk.Unlock()
 
+	// Manifest sender itself is paused.
+	if m.manifest == nil {
+		return nil
+	}
+
 	update := ManifestUpdateMessage{
 		MessageSequence: m.msgSeq,
-	}
-	if !m.paused {
-		update.Manifest = m.manifest
+		Manifest:        *m.manifest,
 	}
 
 	b, err := update.Marshal()
@@ -99,24 +101,5 @@ func (m *ManifestSender) UpdateManifest(manifest *Manifest) {
 	m.lk.Lock()
 	m.manifest = manifest
 	m.msgSeq++
-	m.paused = false
-	m.lk.Unlock()
-}
-
-func (m *ManifestSender) Pause() {
-	m.lk.Lock()
-	if !m.paused {
-		m.paused = true
-		m.msgSeq++
-	}
-	m.lk.Unlock()
-}
-
-func (m *ManifestSender) Resume() {
-	m.lk.Lock()
-	if m.paused {
-		m.paused = false
-		m.msgSeq++
-	}
 	m.lk.Unlock()
 }

test/f3_test.go

@@ -173,7 +173,9 @@ func TestF3DynamicManifest_WithPauseAndRebootstrap(t *testing.T) {
 	env.waitForInstanceNumber(10, 30*time.Second, false)
 	prevInstance := env.nodes[0].currentGpbftInstance()
 
-	env.manifestSender.Pause()
+	prevCopy := *prev
+	prevCopy.Pause = true
+	env.manifestSender.UpdateManifest(&prevCopy)
 
 	env.waitForManifestChange(prev, 30*time.Second)