feat: convert CloseAccount from unary to stream

Signed-off-by: Manfred Touron <94029+moul@users.noreply.github.com>

api/bertyaccount.proto

@@ -15,7 +15,7 @@ service AccountService {
   rpc OpenAccount (OpenAccount.Request) returns (stream OpenAccount.Reply);
 
   // CloseAccount, close the current account
-  rpc CloseAccount (CloseAccount.Request) returns (CloseAccount.Reply);
+  rpc CloseAccount (CloseAccount.Request) returns (stream CloseAccount.Reply);
 
   // ListAccounts retrieves a list of local accounts
   rpc ListAccounts (ListAccounts.Request) returns (ListAccounts.Reply);
@@ -66,7 +66,9 @@ message OpenAccount {
 
 message CloseAccount {
   message Request {}
-  message Reply {}
+  message Reply {
+    berty.protocol.v1.Progress progress = 1;
+  }
 }
 
 message AccountMetadata {

api/protocoltypes.proto

@@ -1319,7 +1319,7 @@ message AttachmentRetrieve {
 message Progress {
   string state = 1;
   string doing = 2;
-  float percent = 3;
+  float progress = 3;
   uint64 completed = 4;
   uint64 total = 5;
   uint64 delay = 6;

go.mod

@@ -115,7 +115,7 @@ require (
 	gorm.io/gorm v1.20.9
 	moul.io/godev v1.7.0
 	moul.io/openfiles v1.2.0
-	moul.io/progress v1.1.0
+	moul.io/progress v1.3.2
 	moul.io/srand v1.6.1
 	moul.io/testman v1.5.0
 	moul.io/u v1.20.0

go/cmd/berty-integration/main.go

@@ -194,7 +194,7 @@ func (i *integration) testbotAdd() error {
 }
 
 func (i *integration) cleanup() {
-	i.manager.Close()
+	i.manager.Close(nil)
 	time.Sleep(100 * time.Millisecond)
 	if i.tempdir != "" {
 		os.RemoveAll(i.tempdir)

go/cmd/berty/main.go

@@ -44,7 +44,7 @@ func runMain(args []string) error {
 		if err != nil {
 			return errcode.TODO.Wrap(err)
 		}
-		defer manager.Close()
+		defer manager.Close(nil)
 	}
 
 	// root command

go/internal/initutil/manager.go

@@ -19,6 +19,7 @@ import (
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"gorm.io/gorm"
+	"moul.io/progress"
 
 	"berty.tech/berty/v2/go/internal/grpcutil"
 	"berty.tech/berty/v2/go/internal/ipfsutil"
@@ -207,54 +208,92 @@ func (m *Manager) RunWorkers() error {
 	return m.workers.Run()
 }
 
-func (m *Manager) Close() error {
+func (m *Manager) Close(prog *progress.Progress) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
+	if prog == nil {
+		prog = progress.New()
+	}
+	prog.AddStep("cancel-context")
+	prog.AddStep("close-client-conn")
+	prog.AddStep("stop-buf-server")
+	prog.AddStep("close-buf-listener")
+	prog.AddStep("stop-grpc-server")
+	prog.AddStep("close-messenger-server")
+	prog.AddStep("close-messenger-protocol-client")
+	prog.AddStep("cleanup-messenger-db")
+	prog.AddStep("close-protocol-server")
+	prog.AddStep("cleanup-ipfs-webui")
+	prog.AddStep("close-ipfs-node")
+	prog.AddStep("close-datastore")
+	prog.AddStep("cleanup-logging")
+
+	prog.Get("cancel-context").SetAsCurrent()
 	if m.ctxCancel != nil {
 		m.ctxCancel()
 	}
 
+	prog.Get("close-client-conn").SetAsCurrent()
 	if m.Node.GRPC.clientConn != nil {
 		m.Node.GRPC.clientConn.Close()
 	}
 
+	prog.Get("stop-buf-server").SetAsCurrent()
 	if m.Node.GRPC.bufServer != nil {
 		m.Node.GRPC.bufServer.Stop()
 	}
 
+	prog.Get("close-buf-listener").SetAsCurrent()
 	if m.Node.GRPC.bufServerListener != nil {
 		m.Node.GRPC.bufServerListener.Close()
 	}
 
+	prog.Get("stop-grpc-server").SetAsCurrent()
 	if m.Node.GRPC.server != nil {
 		m.Node.GRPC.server.Stop()
 	}
 
+	prog.Get("close-messenger-server").SetAsCurrent()
 	if m.Node.Messenger.server != nil {
 		m.Node.Messenger.server.Close()
 	}
+
+	prog.Get("close-messenger-protocol-client").SetAsCurrent()
 	if m.Node.Messenger.protocolClient != nil {
 		m.Node.Messenger.protocolClient.Close()
 	}
+
+	prog.Get("cleanup-messenger-db").SetAsCurrent()
 	if m.Node.Messenger.dbCleanup != nil {
 		m.Node.Messenger.dbCleanup()
 	}
+
+	prog.Get("close-protocol-server").SetAsCurrent()
 	if m.Node.Protocol.server != nil {
 		m.Node.Protocol.server.Close()
 	}
+
+	prog.Get("cleanup-ipfs-webui").SetAsCurrent()
 	if m.Node.Protocol.ipfsWebUICleanup != nil {
 		m.Node.Protocol.ipfsWebUICleanup()
 	}
+
+	prog.Get("close-ipfs-node").SetAsCurrent()
 	if m.Node.Protocol.ipfsNode != nil {
 		m.Node.Protocol.ipfsNode.Close()
 	}
+
+	prog.Get("close-datastore").SetAsCurrent()
 	if m.Datastore.rootDS != nil {
 		m.Datastore.rootDS.Close()
 	}
+
+	prog.Get("cleanup-logging").SetAsCurrent()
 	if m.Logging.cleanup != nil {
 		m.Logging.cleanup()
 	}
+	prog.Get("cleanup-logging").Done()
 
 	return nil
 }

go/internal/initutil/manager_test.go

@@ -92,7 +92,7 @@ func Example_flags() {
 	if err != nil {
 		panic(err)
 	}
-	defer manager.Close()
+	defer manager.Close(nil)
 
 	// configure flags
 	fmt.Println("before", u.JSON(manager.Node.GRPC))
@@ -118,7 +118,7 @@ func Example_noflags() {
 	if err != nil {
 		panic(err)
 	}
-	defer manager.Close()
+	defer manager.Close(nil)
 
 	fs := flag.NewFlagSet("", flag.ExitOnError)
 	manager.SetupLocalProtocolServerFlags(fs)
@@ -167,7 +167,7 @@ func TestTwoConcurrentManagers(t *testing.T) {
 		manager, err := initutil.New(ctx1)
 		require.NoError(t, err)
 		require.NotNil(t, manager)
-		defer manager.Close()
+		defer manager.Close(nil)
 		fs := flag.NewFlagSet("man1", flag.ExitOnError)
 		manager.SetupLoggingFlags(fs)
 		manager.SetupLocalProtocolServerFlags(fs)
@@ -183,7 +183,7 @@ func TestTwoConcurrentManagers(t *testing.T) {
 		manager, err := initutil.New(ctx2)
 		require.NoError(t, err)
 		require.NotNil(t, manager)
-		defer manager.Close()
+		defer manager.Close(nil)
 		fs := flag.NewFlagSet("man2", flag.ExitOnError)
 		manager.SetupLoggingFlags(fs)
 		manager.SetupRemoteNodeFlags(fs)
@@ -260,7 +260,7 @@ func TestLocalProtocolServerAndClient(t *testing.T) {
 	manager, err := initutil.New(ctx)
 	require.NoError(t, err)
 	require.NotNil(t, manager)
-	defer manager.Close()
+	defer manager.Close(nil)
 
 	// configure flags
 	fs := flag.NewFlagSet("test", flag.ExitOnError)
@@ -289,7 +289,7 @@ func TestLocalProtocolServerLeak(t *testing.T) {
 	manager, err := initutil.New(ctx)
 	require.NoError(t, err)
 	require.NotNil(t, manager)
-	defer manager.Close()
+	defer manager.Close(nil)
 
 	// configure flags
 	fs := flag.NewFlagSet("test", flag.ExitOnError)
@@ -310,7 +310,7 @@ func TestCloseOnUninited(t *testing.T) {
 	manager, err := initutil.New(ctx)
 	require.NoError(t, err)
 	require.NotNil(t, manager)
-	manager.Close()
+	manager.Close(nil)
 }
 
 func TestClosingTwice(t *testing.T) {
@@ -332,11 +332,11 @@ func TestClosingTwice(t *testing.T) {
 	_, err = manager.GetLocalProtocolServer()
 	require.NoError(t, err)
 
-	go manager.Close()
-	go manager.Close()
-	go manager.Close()
-	manager.Close()
-	manager.Close()
+	go manager.Close(nil)
+	go manager.Close(nil)
+	go manager.Close(nil)
+	manager.Close(nil)
+	manager.Close(nil)
 }
 
 func TestCloseOpenClose(t *testing.T) {
@@ -361,7 +361,7 @@ func TestRacyClose(t *testing.T) {
 	var wg sync.WaitGroup
 	wg.Add(2)
 	go func() {
-		manager.Close()
+		manager.Close(nil)
 		wg.Done()
 	}()
 	go func() {