Grain Reentrancy

import "github.com/tochemey/goakt/v4/reentrancy"

cfg := reentrancy.New(
    reentrancy.WithMode(reentrancy.AllowAll),
    reentrancy.WithMaxInFlight(10), // optional cap on concurrent requests
)

identity, err := actor.GrainOf[*OrderGrain](ctx, system, "order-42",
    actor.WithGrainReentrancy(cfg))

func (g *OrderGrain) OnReceive(ctx *actor.GrainContext) {
    switch ctx.Message().(type) {
    case *StartBatch:
        if err := ctx.EnableReentrancy(reentrancy.New(reentrancy.WithMode(reentrancy.AllowAll))); err != nil {
            ctx.Err(err)
            return
        }
        ctx.NoErr() // RequestGrain/RequestActor now work, from this message on
    case *BatchDone:
        ctx.DisableReentrancy()
        ctx.NoErr()
    }
}

func (g *FrontGrain) OnReceive(ctx *actor.GrainContext) {
    switch ctx.Message().(type) {
    case *Request:
        reply := ctx.DeferResponse() // take ownership of this Ask's reply

        ctx.RequestGrain(backendID, &Query{}).Then(func(result any, err error) {
            if err != nil {
                reply.Err(err)
                return
            }

            reply.Response(result) // completes the original Ask, on this grain's turn
        })
        // OnReceive returns here; the turn is never blocked
    }
}

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/tochemey/goakt/v4/actor"
	"github.com/tochemey/goakt/v4/reentrancy"
)

// Message types for the request cycle.
type ComputeTotal struct{}                                // client -> OrderGrain
type PriceRequest struct{ Customer *actor.GrainIdentity } // OrderGrain -> PricingGrain
type DiscountRequest struct{}                             // PricingGrain -> OrderGrain (the cycle)
type Discount struct{ Amount int }
type Price struct{ Amount int }

// OrderGrain computes an order total. It answers ComputeTotal by requesting a
// price from PricingGrain; while that request is in flight, PricingGrain
// requests the customer discount back from it.
type OrderGrain struct {
	pricing *actor.GrainIdentity
}

func (g *OrderGrain) OnActivate(ctx context.Context, props *actor.GrainProps) error {
	pricing, err := actor.GrainOf[*PricingGrain](ctx, props.ActorSystem(), "pricing",
		actor.WithGrainReentrancy(reentrancy.New(reentrancy.WithMode(reentrancy.AllowAll))))
	g.pricing = pricing
	return err
}

func (g *OrderGrain) OnReceive(ctx *actor.GrainContext) {
	switch ctx.Message().(type) {
	case *ComputeTotal:
		// 1. Own this Ask's reply beyond the current turn.
		reply := ctx.DeferResponse()

		ctx.RequestGrain(g.pricing, &PriceRequest{Customer: ctx.Self()}).Then(func(result any, err error) {
			if err != nil {
				reply.Err(err)
				return
			}

			// 5. Complete the original AskGrain from the continuation.
			reply.Response(result.(*Price))
		})
	case *DiscountRequest:
		// 3. Arrives from PricingGrain while our own request to it is still in
		// flight. AllowAll lets this turn run instead of deadlocking the cycle.
		ctx.Response(&Discount{Amount: 10})
	default:
		ctx.Unhandled()
	}
}

func (g *OrderGrain) OnDeactivate(ctx context.Context, props *actor.GrainProps) error {
	return nil
}

// PricingGrain prices an order, but needs the customer discount from the
// requesting OrderGrain before it can answer.
type PricingGrain struct{}

func (g *PricingGrain) OnActivate(ctx context.Context, props *actor.GrainProps) error {
	return nil
}

func (g *PricingGrain) OnReceive(ctx *actor.GrainContext) {
	switch msg := ctx.Message().(type) {
	case *PriceRequest:
		// 2. Still owing the reply to OrderGrain, request it back.
		reply := ctx.DeferResponse()

		ctx.RequestGrain(msg.Customer, &DiscountRequest{}).Then(func(result any, err error) {
			if err != nil {
				reply.Err(err)
				return
			}

			// 4. Answer OrderGrain's request with the discounted price.
			reply.Response(&Price{Amount: 100 - result.(*Discount).Amount})
		})
	default:
		ctx.Unhandled()
	}
}

func (g *PricingGrain) OnDeactivate(ctx context.Context, props *actor.GrainProps) error {
	return nil
}

func main() {
	ctx := context.Background()

	system, err := actor.NewActorSystem("grain-reentrancy-demo", actor.WithLoggingDisabled())
	if err != nil {
		panic(err)
	}

	if err := system.Start(ctx); err != nil {
		panic(err)
	}
	defer system.Stop(ctx)

	order, err := actor.GrainOf[*OrderGrain](ctx, system, "order-42",
		actor.WithGrainReentrancy(reentrancy.New(reentrancy.WithMode(reentrancy.AllowAll))))
	if err != nil {
		panic(err)
	}

	total, err := system.AskGrain(ctx, order, &ComputeTotal{}, 3*time.Second)
	if err != nil {
		panic(err)
	}

	fmt.Println("total:", total.(*Price).Amount) // total: 90
}