Improve lifecycle management

cmd/f3/manifest.go

@@ -14,6 +14,7 @@
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/urfave/cli/v2"
+	"golang.org/x/sync/errgroup"
 	"golang.org/x/xerrors"
 )
 
@@ -137,6 +138,8 @@
 			return xerrors.Errorf("initializing libp2p host: %w", err)
 		}
 
+		defer func() { _ = host.Close() }()
+
 		// Connect to all bootstrap addresses once. This should be sufficient to build
 		// the pubsub mesh, if not then we need to periodically re-connect and/or pull in
 		// the Lotus bootstrapping, which includes DHT connectivity.
@@ -171,15 +174,15 @@
 		}
 
 		manifestPath := c.String("manifest")
-		loadManifestAndVersion := func() (manifest.Manifest, manifest.Version, error) {
+		loadManifestAndVersion := func() (*manifest.Manifest, manifest.Version, error) {
 
 			m, err := loadManifest(manifestPath)
 			if err != nil {
-				return manifest.Manifest{}, "", xerrors.Errorf("loading manifest: %w", err)
+				return nil, "", xerrors.Errorf("loading manifest: %w", err)
 			}
 			version, err := m.Version()
 			if err != nil {
-				return manifest.Manifest{}, "", xerrors.Errorf("versioning manifest: %w", err)
+				return nil, "", xerrors.Errorf("versioning manifest: %w", err)
 			}
 			return m, version, nil
 		}
@@ -200,46 +203,49 @@
 		}
 		_, _ = fmt.Fprintf(c.App.Writer, "Started manifest sender with version: %s\n", manifestVersion)
 
-		go func() {
-			sender.Start(c.Context)
-		}()
-		defer func() {
-			sender.Stop()
-			_ = host.Close()
-		}()
-
-		checkTicker := time.NewTicker(c.Duration("checkInterval"))
-		for c.Context.Err() == nil {
-			select {
-			case <-c.Context.Done():
-				return c.Context.Err()
-			case <-checkTicker.C:
-				if nextManifest, nextManifestVersion, err := loadManifestAndVersion(); err != nil {
-					_, _ = fmt.Fprintf(c.App.ErrWriter, "Failed reload manifest: %v\n", err)
-				} else if manifestVersion != nextManifestVersion {
-					_, _ = fmt.Fprintf(c.App.Writer, "Loaded manifest with version: %s\n", nextManifestVersion)
-					sender.UpdateManifest(nextManifest)
-					manifestVersion = nextManifestVersion
+		checkInterval := c.Duration("checkInterval")
+
+		errgrp, ctx := errgroup.WithContext(c.Context)
+		errgrp.Go(func() error { return sender.Run(ctx) })
+		errgrp.Go(func() error {
+			checkTicker := time.NewTicker(checkInterval)
+			defer checkTicker.Stop()
+
+			for ctx.Err() == nil {
+				select {
+				case <-ctx.Done():
+					return nil
+				case <-checkTicker.C:
+					if nextManifest, nextManifestVersion, err := loadManifestAndVersion(); err != nil {
+						_, _ = fmt.Fprintf(c.App.ErrWriter, "Failed reload manifest: %v\n", err)
+					} else if manifestVersion != nextManifestVersion {
+						_, _ = fmt.Fprintf(c.App.Writer, "Loaded manifest with version: %s\n", nextManifestVersion)
+						sender.UpdateManifest(nextManifest)
+						manifestVersion = nextManifestVersion
+					}
 				}
 			}
-		}
-		return nil
+
+			return nil
+		})
+
+		return errgrp.Wait()
 	},
 }
 
-func getManifest(c *cli.Context) (manifest.Manifest, error) {
+func getManifest(c *cli.Context) (*manifest.Manifest, error) {
 	manifestPath := c.String("manifest")
 	return loadManifest(manifestPath)
 }
 
-func loadManifest(path string) (manifest.Manifest, error) {
+func loadManifest(path string) (*manifest.Manifest, error) {
 	f, err := os.Open(path)
 	if err != nil {
-		return manifest.Manifest{}, xerrors.Errorf("opening %s to load manifest: %w", path, err)
+		return nil, xerrors.Errorf("opening %s to load manifest: %w", path, err)
 	}
 	defer f.Close()
 	var m manifest.Manifest
 
 	err = m.Unmarshal(f)
-	return m, err
+	return &m, err
 }

cmd/f3/run.go

@@ -121,20 +121,18 @@
 			return xerrors.Errorf("creating module: %w", err)
 		}
 
-		mprovider.SetManifestChangeCallback(f3.ManifestChangeCallback(module))
 		go runMessageSubscription(ctx, module, gpbft.ActorID(id), signingBackend)
 
-		return module.Run(ctx)
+		if err := module.Start(ctx); err != nil {
+			return nil
+		}
+		<-ctx.Done()
+		return module.Stop(context.Background())
 	},
 }
 
 func runMessageSubscription(ctx context.Context, module *f3.F3, actorID gpbft.ActorID, signer gpbft.Signer) {
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
+	for ctx.Err() == nil {
 
 		ch := make(chan *gpbft.MessageBuilder, 4)
 		module.SubscribeForMessagesToSign(ch)

ec/powerdelta.go

@@ -0,0 +1,29 @@
+package ec
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/filecoin-project/go-f3/certs"
+	"github.com/filecoin-project/go-f3/gpbft"
+)
+
+func WithModifiedPower(backend Backend, delta []certs.PowerTableDelta) Backend {
+	return &withModifiedPower{
+		Backend: backend,
+		delta:   delta,
+	}
+}
+
+type withModifiedPower struct {
+	Backend
+	delta []certs.PowerTableDelta
+}
+
+func (b *withModifiedPower) GetPowerTable(ctx context.Context, ts gpbft.TipSetKey) (gpbft.PowerEntries, error) {
+	pt, err := b.Backend.GetPowerTable(ctx, ts)
+	if err != nil {
+		return nil, fmt.Errorf("getting power table: %w", err)
+	}
+	return certs.ApplyPowerTableDiffs(pt, b.delta)
+}