Bug Maybe: Calling Respond while using the Persistence Mixin

[chriskolenko]

Hi all,

I'm not sure if this is a bug or not.

I got a deadletter when using the Respond after persisting a messaging.

	if !a.Recovering() {
		a.PersistReceive(msg)
		ctx.Respond(..new..message..)
	}

I only get the issue when PersistReceive will Request a Snapshot.

After reversing the order

	if !a.Recovering() {
		ctx.Respond(..new..message..)
		a.PersistReceive(msg)
	}

Everything works as expected.

[ljluestc]

package main

import (
	"fmt"
	"log"

	"github.com/AsynkronIT/protoactor-go/actor"
	"github.com/AsynkronIT/protoactor-go/persistence"
)

// Message represents an incoming message to the actor
type Message struct {
	Content string
}

// Response represents the actor's response
type Response struct {
	Reply string
}

// MyActor is the persistent actor
type MyActor struct {
	persistence.Mixin // Embeds persistence behavior
	state             string
}

// Receive handles incoming messages
func (a *MyActor) Receive(ctx actor.Context) {
	switch msg := ctx.Message().(type) {
	case *persistence.ReplayComplete:
		// Recovery is complete, no action needed
		log.Println("Replay complete")
	case *Message:
		if !a.Recovering() {
			// Respond first to avoid dead letters during snapshot
			ctx.Respond(&Response{Reply: fmt.Sprintf("Processed: %s", msg.Content)})
			// Persist the message after responding
			a.PersistReceive(msg)
			// Update state
			a.state = msg.Content
			// Optionally request a snapshot (simulate the issue)
			if len(a.state)%10 == 0 { // Example condition for snapshot
				a.RequestSnapshot()
			}
		}
	case *persistence.Event:
		// Handle persisted events during recovery
		if eventMsg, ok := msg.Data.(*Message); ok {
			a.state = eventMsg.Content
			log.Printf("Recovered state: %s", a.state)
		}
	case *persistence.Snapshot:
		// Handle snapshot recovery
		if snapshotState, ok := msg.Data.(string); ok {
			a.state = snapshotState
			log.Printf("Recovered snapshot: %s", a.state)
		}
	}
}

func main() {
	// Set up the actor system
	system := actor.NewActorSystem()

	// Configure persistence provider (in-memory for demo)
	provider := persistence.NewInMemoryProvider()

	// Actor props with persistence
	props := actor.PropsFromProducer(func() actor.Actor {
		return &MyActor{state: "initial"}
	}, actor.WithReceiverMiddleware(persistence.Using(provider)))

	// Spawn the actor
	pid := system.Root.Spawn(props)

	// Send a test message
	response, err := system.Root.RequestFuture(pid, &Message{Content: "Hello"}, -1).Result()
	if err != nil {
		log.Fatalf("Failed to get response: %v", err)
	}
	if resp, ok := response.(*Response); ok {
		log.Printf("Received response: %s", resp.Reply)
	}

	// Simulate multiple messages to trigger snapshot
	for i := 0; i < 12; i++ {
		response, err = system.Root.RequestFuture(pid, &Message{Content: fmt.Sprintf("Message-%d", i)}, -1).Result()
		if err != nil {
			log.Printf("Error on message %d: %v", i, err)
		} else if resp, ok := response.(*Response); ok {
			log.Printf("Response for message %d: %s", i, resp.Reply)
		}
	}

	// Wait to observe output
	fmt.Scanln()
}