Fixing race condition in LeaserCheckpointer where it can fail with a …

…ContainerAlreadyExists error #253

There's a slight race condition between checking the store exists and the container being created.
    
We can handle it easily if we just allow ContainerAlreadyExists to be considered successful. Also, since the storage tests were failing (unrelated to my change) in race detection I also fixed that as well.

Fixes #252
Fixes #225

changelog.md

@@ -1,5 +1,9 @@
 # Change Log
 
+## `v3.3.17`
+
+- Fixing issue where the LeaserCheckpointer could fail with a "ContainerAlreadyExists" error. (#253)
+
 ## `v3.3.16`
 
 - Exporting a subset of AMQP message properties for the Dapr project.

eng/azure-pipelines.yml

@@ -41,7 +41,7 @@ jobs:
           go get github.com/AlekSi/gocov-xml
           go get -u github.com/matm/gocov-html
           go get -u golang.org/x/lint/golint
-          go get github.com/fzipp/gocyclo/cmd/gocyclo
+          go get github.com/fzipp/gocyclo/cmd/gocyclo@v0.3.1
         workingDirectory: '$(sdkPath)'
         displayName: 'Install Dependencies'
       - script: |

eng/integration-tests.yml

@@ -29,7 +29,7 @@ steps:
       go get github.com/axw/gocov/gocov
       go get github.com/AlekSi/gocov-xml
       go get -u github.com/matm/gocov-html
-      go get github.com/fzipp/gocyclo/cmd/gocyclo
+      go get github.com/fzipp/gocyclo/cmd/gocyclo@v0.3.1
       go get golang.org/x/lint/golint
     displayName: 'Install Dependencies'
   - script: |
@@ -47,10 +47,10 @@ steps:
       gocov-html < coverage.json > coverage.html
     displayName: 'Run Integration Tests'
     env:
-      ARM_SUBSCRIPTION_ID: $(go-live-azure-subscription-id)
-      ARM_CLIENT_ID: $(go-live-eh-azure-client-id)
-      ARM_CLIENT_SECRET: $(go-live-eh-azure-client-secret)
-      ARM_TENANT_ID: $(go-live-tenant-id)
+      ARM_SUBSCRIPTION_ID: $(azure-subscription-id)
+      ARM_CLIENT_ID: $(aad-azure-sdk-test-client-id)
+      ARM_CLIENT_SECRET: $(aad-azure-sdk-test-client-secret)
+      ARM_TENANT_ID: $(aad-azure-sdk-test-tenant-id)
 
   - task: PublishTestResults@2
     inputs:

eph/leasedReceiver.go

@@ -31,7 +31,7 @@ import (
 
 	"github.com/devigned/tab"
 
-	"github.com/Azure/azure-event-hubs-go/v3"
+	eventhub "github.com/Azure/azure-event-hubs-go/v3"
 )
 
 type (
@@ -58,11 +58,10 @@ func (lr *leasedReceiver) Run(ctx context.Context) error {
 	epoch := lr.lease.GetEpoch()
 	lr.dlog(ctx, "running...")
 
-	go func() {
-		ctx, done := context.WithCancel(context.Background())
-		lr.done = done
-		lr.periodicallyRenewLease(ctx)
-	}()
+	renewLeaseCtx, cancelRenewLease := context.WithCancel(context.Background())
+	lr.done = cancelRenewLease
+
+	go lr.periodicallyRenewLease(renewLeaseCtx)
 
 	opts := []eventhub.ReceiveOption{eventhub.ReceiveWithEpoch(epoch)}
 	if lr.processor.consumerGroup != "" {

storage/storage.go

@@ -30,14 +30,15 @@ import (
 	"encoding/json"
 	"errors"
 	"io/ioutil"
+	"net/http"
 	"net/url"
 	"sync"
 	"time"
 
 	"github.com/Azure/azure-amqp-common-go/v3/uuid"
 	"github.com/devigned/tab"
 
-	"github.com/Azure/azure-event-hubs-go/v3"
+	eventhub "github.com/Azure/azure-event-hubs-go/v3"
 	"github.com/Azure/azure-event-hubs-go/v3/eph"
 	"github.com/Azure/azure-event-hubs-go/v3/persist"
 
@@ -144,20 +145,22 @@ func (sl *LeaserCheckpointer) StoreExists(ctx context.Context) (bool, error) {
 	span, ctx := startConsumerSpanFromContext(ctx, "storage.LeaserCheckpointer.StoreExists")
 	defer span.End()
 
-	opts := azblob.ListContainersSegmentOptions{
-		Prefix: sl.containerName,
-	}
-	res, err := sl.serviceURL.ListContainersSegment(ctx, azblob.Marker{}, opts)
-	if err != nil {
-		return false, err
+	containerURL := sl.serviceURL.NewContainerURL(sl.containerName)
+	_, err := containerURL.GetProperties(ctx, azblob.LeaseAccessConditions{})
+
+	if err == nil {
+		return true, nil
 	}
 
-	for _, container := range res.ContainerItems {
-		if container.Name == sl.containerName {
-			return true, nil
+	var respErr azblob.ResponseError
+
+	if errors.As(err, &respErr) {
+		if respErr.Response().StatusCode == http.StatusNotFound {
+			return false, nil
 		}
 	}
-	return false, nil
+
+	return false, err
 }
 
 // EnsureStore creates the container if it does not exist
@@ -175,9 +178,21 @@ func (sl *LeaserCheckpointer) EnsureStore(ctx context.Context) error {
 	if !ok {
 		containerURL := sl.serviceURL.NewContainerURL(sl.containerName)
 		_, err := containerURL.Create(ctx, azblob.Metadata{}, azblob.PublicAccessNone)
+
 		if err != nil {
-			return err
+			var storageErr azblob.StorageError
+
+			if errors.As(err, &storageErr) {
+				// we're okay if the container has been created - we're basically racing against
+				// other LeaserCheckpointers.
+				if storageErr.ServiceCode() != azblob.ServiceCodeContainerAlreadyExists {
+					return err
+				}
+			} else {
+				return err
+			}
 		}
+
 		sl.containerURL = &containerURL
 	}
 	return nil

storage/storage_test.go

@@ -25,13 +25,15 @@ package storage
 import (
 	"context"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/Azure/azure-amqp-common-go/v3/aad"
 	"github.com/Azure/azure-storage-blob-go/azblob"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/Azure/azure-event-hubs-go/v3/eph"
+	"github.com/Azure/azure-event-hubs-go/v3/internal/test"
 )
 
 const (
@@ -64,6 +66,35 @@ func (ts *testSuite) TestLeaserStoreCreation() {
 	ts.True(exists)
 }
 
+func (ts *testSuite) TestLeaserStoreCreationConcurrent() {
+	wg := sync.WaitGroup{}
+
+	containerName := test.RandomString("concurrent-container", 4)
+
+	// do a simple test that ensures we don't die just because we raced with
+	// other leasers to create the storage container.
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+
+		go func(i int) {
+			defer wg.Done()
+
+			leaser, _ := ts.newLeaserWithContainerName(containerName)
+
+			err := leaser.EnsureStore(context.Background())
+			ts.Require().NoError(err)
+		}(i)
+	}
+
+	wg.Wait()
+
+	leaser, del := ts.newLeaserWithContainerName(containerName)
+	defer del()
+	exists, err := leaser.StoreExists(context.Background())
+	ts.NoError(err)
+	ts.True(exists)
+}
+
 func (ts *testSuite) TestLeaserLeaseEnsure() {
 	leaser, del := ts.leaserWithEPH()
 	defer del()
@@ -189,6 +220,10 @@ func (ts *testSuite) leaserWithEPH() (*LeaserCheckpointer, func()) {
 
 func (ts *testSuite) newLeaser() (*LeaserCheckpointer, func()) {
 	containerName := strings.ToLower(ts.RandomName("stortest", 4))
+	return ts.newLeaserWithContainerName(containerName)
+}
+
+func (ts *testSuite) newLeaserWithContainerName(containerName string) (*LeaserCheckpointer, func()) {
 	cred, err := NewAADSASCredential(ts.SubscriptionID, ts.ResourceGroupName, ts.AccountName, containerName, AADSASCredentialWithEnvironmentVars())
 	ts.Require().NoError(err)
 	leaser, err := NewStorageLeaserCheckpointer(cred, ts.AccountName, containerName, ts.Env)

version.go

@@ -2,5 +2,5 @@ package eventhub
 
 const (
 	// Version is the semantic version number
-	Version = "3.3.13"
+	Version = "3.3.17"
 )