Chained Futures do not unblock selects if "source" future IsReady

[Groxx]

package experiments

import (
    "testing"
    "time"

    "github.com/stretchr/testify/assert"
    "go.uber.org/cadence/activity"
    "go.uber.org/cadence/testsuite"
    "go.uber.org/cadence/workflow"
)

func init() {
    workflow.Register(selectorWork)
    activity.Register(selectorAct)
}

func selectorWork(ctx workflow.Context, sync bool) error {
    ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        ScheduleToStartTimeout: time.Minute,
        StartToCloseTimeout:    time.Minute,
    })

    fut, set := workflow.NewFuture(ctx)
    orig := workflow.ExecuteActivity(ctx, selectorAct, 5)

    if sync {
        orig.Get(ctx, nil)
    }
    set.Chain(orig)

    s := workflow.NewSelector(ctx)
    s.AddFuture(fut, func(f workflow.Future) {
        // nothing to do, unblocking will pass the test
    })
    s.Select(ctx)
    return nil
}

func selectorAct(arg int) (int, error) {
    return arg, nil
}

func TestSelector(t *testing.T) {
    runner := func(block bool) func(t *testing.T) {
        return func(t *testing.T) {
            s := new(testsuite.WorkflowTestSuite)
            env := s.NewTestWorkflowEnvironment()

            env.ExecuteWorkflow(selectorWork, block)

            assert.True(t, env.IsWorkflowCompleted())
            assert.NoError(t, env.GetWorkflowError())
        }
    }

    t.Run("async chains work", runner(false))
    t.Run("sync chains never unblock", runner(true))
}

If you comment out one of those two, you can see that async passes ~immediately, while sync times out.

Cause appears to be that the IsReady branch doesn't close the related channel? https://github.com/uber-go/cadence-client/blob/9c4380588cc1452dc0e84d672cb1f3dd758ae67a/internal/internal_workflow.go#L329

func (f *futureImpl) Chain(future Future) {
    if f.ready {
        panic("already set")
    }

    ch, ok := future.(asyncFuture)
    if !ok {
        panic("cannot chain Future that wasn't created with workflow.NewFuture")
    }
    if !ch.IsReady() {
        ch.ChainFuture(f)
        return
    }
    val, err := ch.GetValueAndError()
    f.value = val
    f.err = err
    f.ready = true
    return
}

Compare to e.g. Set:

func (f *futureImpl) Set(value interface{}, err error) {
    if f.ready {
        panic("already set")
    }
    f.value = value
    f.err = err
    f.ready = true
    f.channel.Close()
    for _, ch := range f.chained {
        ch.Set(f.value, f.err)
    }
}

[kotenko-anton]

I've faced with the same issue and came to the same conclusions. Workaround it with

newFuture, settable := workflow.NewFuture(ctx)
...........................
selector.AddFuture(newFuture, func(f workflow.Future) {.......})
...........................
var value interface{}
err := future.Get(ctx, &value)
if err != nil {
   settable.SetError(err)
} else {
  settable.SetValue(value)
}
...........................
selector.AddFuture()

Do you have plans to fix it? or if you are able to confirm that it is really source of problem I can prepare pull request that fixes the problem.