Fix current exchange for remote signers

gossip3/actors/currentstateexchange.go

@@ -42,14 +42,20 @@ func NewCurrentStateExchangeProps(config *CurrentStateExchangeConfig) *actor.Pro
 
 func (e *CurrentStateExchange) Receive(context actor.Context) {
 	switch msg := context.Message().(type) {
+	case *messages.DoCurrentStateExchange:
+		context.Request(msg.Destination, &messages.RequestCurrentStateSnapshot{})
 	case *messages.RequestCurrentStateSnapshot:
 		e.handleRequestCurrentStateSnapshot(context, msg)
 	case *messages.ReceiveCurrentStateSnapshot:
-		e.handleReceiveCurrentStateSnapshot(context, msg)
+		e.gzipImport(context, msg.Payload)
 	}
 }
 
 func (e *CurrentStateExchange) handleRequestCurrentStateSnapshot(context actor.Context, msg *messages.RequestCurrentStateSnapshot) {
+	if context.Sender() == nil {
+		panic("RequestCurrentStateSnapshot requires a Sender")
+	}
+
 	gzippedBytes := e.gzipExport()
 
 	if len(gzippedBytes) == 0 {
@@ -60,18 +66,7 @@ func (e *CurrentStateExchange) handleRequestCurrentStateSnapshot(context actor.C
 		Payload: gzippedBytes,
 	}
 
-	_, err := context.RequestFuture(extmsgs.FromActorPid(msg.Destination), payload, CURRENT_STATE_EXCHANGE_TIMEOUT).Result()
-	if err != nil {
-		panic(fmt.Sprintf("exchange with %v failed: %v", msg.Destination, err))
-	}
-}
-
-func (e *CurrentStateExchange) handleReceiveCurrentStateSnapshot(context actor.Context, msg *messages.ReceiveCurrentStateSnapshot) {
-	var responseBytes []byte
-	e.gzipImport(context, msg.Payload)
-	context.Respond(&messages.ReceiveCurrentStateSnapshot{
-		Payload: responseBytes,
-	})
+	context.Request(context.Sender(), payload)
 }
 
 func (e *CurrentStateExchange) gzipExport() []byte {
@@ -114,7 +109,7 @@ func (e *CurrentStateExchange) gzipImport(context actor.Context, payload []byte)
 	}
 	defer reader.Close()
 
-	e.Log.Debugw("gzipImport from %v started", context.Sender())
+	e.Log.Debugf("gzipImport from %v started", context.Sender())
 
 	bytesLeft := true
 
@@ -146,5 +141,5 @@ func (e *CurrentStateExchange) gzipImport(context actor.Context, payload []byte)
 		bytesLeft = buf.Len() > 0
 	}
 
-	e.Log.Debugw("gzipImport from %v processed %d keys", context.Sender(), wroteCount)
+	e.Log.Debugf("gzipImport from %v processed %d keys", context.Sender(), wroteCount)
 }

gossip3/actors/tupelo.go

@@ -70,8 +70,6 @@ func (tn *TupeloNode) Receive(context actor.Context) {
 		tn.handleNewTransaction(context)
 	case *messages.RequestCurrentStateSnapshot:
 		context.Forward(tn.currentStateExchangeActor)
-	case *messages.ReceiveCurrentStateSnapshot:
-		context.Forward(tn.currentStateExchangeActor)
 	}
 }
 
@@ -249,8 +247,6 @@ func (tn *TupeloNode) handleStarted(context actor.Context) {
 			}
 		}
 
-		context.Send(otherSigner, &messages.RequestCurrentStateSnapshot{
-			Destination: extmsgs.ToActorPid(tn.currentStateExchangeActor),
-		})
+		context.Send(tn.currentStateExchangeActor, &messages.DoCurrentStateExchange{Destination: otherSigner})
 	}
 }

gossip3/messages/internal.go

@@ -12,7 +12,6 @@ func init() {
 }
 
 type RequestCurrentStateSnapshot struct {
-	Destination *extmsgs.ActorPID
 }
 
 func (RequestCurrentStateSnapshot) TypeCode() int8 {

gossip3/messages/messages.go

@@ -3,6 +3,7 @@ package messages
 import (
 	"fmt"
 
+	"github.com/AsynkronIT/protoactor-go/actor"
 	extmsgs "github.com/quorumcontrol/tupelo-go-sdk/gossip3/messages"
 	"github.com/quorumcontrol/tupelo-go-sdk/gossip3/types"
 	"github.com/quorumcontrol/tupelo-go-sdk/tracing"
@@ -84,3 +85,7 @@ type GzipImport struct {
 type ImportCurrentState struct {
 	CurrentState *extmsgs.CurrentState
 }
+
+type DoCurrentStateExchange struct {
+	Destination *actor.PID
+}