Leadership: re-reconcile on failed key read (#68)

In the event of failed reading of other leadership keys, the leader will
continue to wait for the next informer reconcile instead of hard
erroring out.

This is useful for events whereby the leadership key space values are
not readable because they have been written by another cron who has a
different value API.

Signed-off-by: joshvanl <me@joshvanl.dev>

internal/leadership/elector/elector.go

@@ -128,7 +128,11 @@ func (e *Elector) Elect(ctx context.Context) (context.Context, *Elected, error)
 	for {
 		ldata, ok, err = e.quorumReconcile(ctx, resp)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to reconcile to leadership quorum: %w", err)
+			e.log.Error(err, "failed to reconcile to leadership quorum")
+		}
+
+		if ctx.Err() != nil {
+			return nil, nil, err
 		}
 
 		if ok {
@@ -191,7 +195,11 @@ func (e *Elector) Reelect(ctx context.Context) (context.Context, *Elected, error
 	for {
 		ldata, ok, err = e.quorumReconcile(ctx, resp)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to reconcile to leadership quorum: %w", err)
+			e.log.Error(err, "failed to reconcile to leadership quorum")
+		}
+
+		if ctx.Err() != nil {
+			return nil, nil, err
 		}
 
 		if ok {

tests/suite/leadership_test.go

@@ -13,8 +13,10 @@ import (
 	"time"
 
 	"github.com/dapr/kit/ptr"
+	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/diagridio/go-etcd-cron/api"
@@ -226,3 +228,55 @@ func Test_leadership_scaledown(t *testing.T) {
 
 	assert.Len(t, instanceCalled, 3)
 }
+
+func Test_leadership_wait_free(t *testing.T) {
+	t.Parallel()
+
+	client := etcd.EmbeddedBareClient(t)
+	opts := cron.Options{
+		Client: client,
+		Log:    logr.Discard(),
+		ID:     "123",
+		TriggerFn: func(_ context.Context, req *api.TriggerRequest) *api.TriggerResponse {
+			return &api.TriggerResponse{Result: api.TriggerResponseResult_SUCCESS}
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	_, err := client.Put(ctx, "leadership/456", "123")
+	require.NoError(t, err)
+
+	cr, err := cron.New(opts)
+	require.NoError(t, err)
+
+	errCh := make(chan error)
+	go func() { errCh <- cr.Run(ctx) }()
+
+	t.Cleanup(func() {
+		cancel()
+		select {
+		case <-time.After(time.Second * 5):
+			t.Fatal("timeout waiting for cron return")
+		case err := <-errCh:
+			require.NoError(t, err)
+		}
+	})
+
+	select {
+	case err := <-errCh:
+		t.Fatal(err)
+	case <-time.After(time.Second * 5):
+	}
+
+	_, err = client.Delete(ctx, "leadership/456")
+	require.NoError(t, err)
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.True(c, cr.IsElected())
+	}, time.Second*10, time.Millisecond*10)
+
+	resp, err := client.Get(ctx, "", clientv3.WithPrefix())
+	require.NoError(t, err)
+	require.Len(t, resp.Kvs, 1)
+}