Repository: rollbar/rollbar-go
Pull request #100 (2022-07-21 06:19:59)

added recovering in case of activity on a closed channel
## Description of the change
1. Gracefully recover when sending/receiving on the closed channel (protecting from panic)
issue# : #90 
3. Fixed race conditions (also protecting from possible panic)
## Type of change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] Maintenance
- [ ] New release

## Related issues

> Shortcut stories and GitHub issues (delete irrelevant)

- Fix [SC-]
- Fix #1

## Checklists

### Development

- [x] Lint rules pass locally
- [x] The code changed/added as part of this pull request has been covered with tests
- [x] All tests related to the changed code pass in development

### Code review

- [x] This pull request has a descriptive title and information useful to a reviewer. There may be a screenshot or screencast attached
- [x] "Ready for review" label attached to the PR and reviewers assigned
- [ ] Issue from task tracker has a link to this pull request
- [x] Changes have been reviewed by at least one other engineer



Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 85b438b..ea7bc01 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -33,4 +33,4 @@ jobs:
           node-version: ${{ matrix.go-version }}
 
       - name: Run tests
-        run: go test
+        run: go test --race
diff --git a/async_transport.go b/async_transport.go
index 10fb109..747fd65 100644
--- a/async_transport.go
+++ b/async_transport.go
@@ -1,6 +1,8 @@
 package rollbar
 
 import (
+	"fmt"
+	"runtime"
 	"sync"
 	"time"
 )
@@ -21,6 +23,16 @@ type payload struct {
 	retriesLeft int
 }
 
+func isClosed(ch chan payload) bool {
+	if len(ch) == 0 {
+		select {
+		case _, ok := <-ch:
+			return !ok
+		}
+	}
+	return false
+}
+
 // NewAsyncTransport builds an asynchronous transport which sends data to the Rollbar API at the
 // specified endpoint using the given access token. The channel is limited to the size of the input
 // buffer argument.
@@ -38,6 +50,18 @@ func NewAsyncTransport(token string, endpoint string, buffer int) *AsyncTranspor
 	}
 
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				pc, _, _, _ := runtime.Caller(4)
+				fnName := runtime.FuncForPC(pc).Name()
+				if isClosed(transport.bodyChannel) {
+					fmt.Println(fnName, "recover: channel is closed")
+				} else {
+					fmt.Println(fnName, "recovered:", r)
+				}
+			}
+		}()
+
 		for p := range transport.bodyChannel {
 			elapsedTime := time.Now().Sub(transport.startTime).Seconds()
 			if elapsedTime < 0 || elapsedTime >= 60 {
@@ -81,7 +105,20 @@ func NewAsyncTransport(token string, endpoint string, buffer int) *AsyncTranspor
 
 // Send the body to Rollbar if the channel is not currently full.
 // Returns ErrBufferFull if the underlying channel is full.
-func (t *AsyncTransport) Send(body map[string]interface{}) error {
+func (t *AsyncTransport) Send(body map[string]interface{}) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			pc, _, _, _ := runtime.Caller(4)
+			fnName := runtime.FuncForPC(pc).Name()
+			if _, ok := err.(*ErrBufferFull); !ok && isClosed(t.bodyChannel) {
+				fmt.Println(fnName, "recover: channel is closed")
+				t.waitGroup.Done()
+				err = ErrChannelClosed{}
+			} else {
+				fmt.Println(fnName, "recovered:", r)
+			}
+		}
+	}()
 	if len(t.bodyChannel) < t.Buffer {
 		t.waitGroup.Add(1)
 		p := payload{
@@ -90,7 +127,7 @@ func (t *AsyncTransport) Send(body map[string]interface{}) error {
 		}
 		t.bodyChannel <- p
 	} else {
-		err := ErrBufferFull{}
+		err = ErrBufferFull{}
 		rollbarError(t.Logger, err.Error())
 		if t.PrintPayloadOnError {
 			writePayloadToStderr(t.Logger, body)
diff --git a/base_transport.go b/base_transport.go
index 33ec81f..fb6a120 100644
--- a/base_transport.go
+++ b/base_transport.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"sync"
 	"time"
 )
 
@@ -33,6 +34,7 @@ type baseTransport struct {
 
 	perMinCounter int
 	startTime     time.Time
+	lock          sync.RWMutex
 }
 
 // SetToken updates the token to use for future API requests.
@@ -47,12 +49,16 @@ func (t *baseTransport) SetEndpoint(endpoint string) {
 
 // SetItemsPerMinute sets the max number of items to send in a given minute
 func (t *baseTransport) SetItemsPerMinute(itemsPerMinute int) {
+	t.lock.Lock()
 	t.ItemsPerMinute = itemsPerMinute
+	t.lock.Unlock()
 }
 
 // SetLogger updates the logger that this transport uses for reporting errors that occur while
 // processing items.
 func (t *baseTransport) SetLogger(logger ClientLogger) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
 	t.Logger = logger
 }
 
@@ -99,6 +105,8 @@ func (t *baseTransport) clientPost(body io.Reader) (resp *http.Response, err err
 // value is true then the caller could call this function again with the same input and possibly
 // see a non-error response.
 func (t *baseTransport) post(body map[string]interface{}) (bool, error) {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
 	if len(t.Token) == 0 {
 		rollbarError(t.Logger, "empty token")
 		return false, nil
@@ -130,6 +138,8 @@ func (t *baseTransport) post(body map[string]interface{}) (bool, error) {
 }
 
 func (t *baseTransport) shouldSend() bool {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
 	if t.ItemsPerMinute > 0 && t.perMinCounter >= t.ItemsPerMinute {
 		rollbarError(t.Logger, fmt.Sprintf("item per minute limit reached: %d occurences, "+
 			"ignoring errors until timeout", t.perMinCounter))
diff --git a/errors.go b/errors.go
index 293d0e2..f28e8b6 100644
--- a/errors.go
+++ b/errors.go
@@ -21,3 +21,12 @@ type ErrBufferFull struct{}
 func (e ErrBufferFull) Error() string {
 	return "buffer full, dropping error on the floor"
 }
+
+// ErrChannelClosed is an error which is returned when the asynchronous transport is used and the
+// channel used for buffering items for sending to Rollbar is already closed
+type ErrChannelClosed struct{}
+
+// Error implements the error interface.
+func (e ErrChannelClosed) Error() string {
+	return "channel is closed"
+}


Test patch:
diff --git a/async_transport_test.go b/async_transport_test.go
index 192372a..a32617e 100644
--- a/async_transport_test.go
+++ b/async_transport_test.go
@@ -18,23 +18,34 @@ func TestAsyncTransportSend(t *testing.T) {
 }
 
 func TestAsyncTransportSendFull(t *testing.T) {
-	transport := NewAsyncTransport("", "", 1)
+	transport := NewAsyncTransport("", "", 0)
 	transport.SetLogger(&SilentClientLogger{})
 	body := map[string]interface{}{
 		"hello": "world",
 	}
 
-	transport.Send(body)
 	result := transport.Send(body)
 	if result == nil {
 		t.Error("Expected to receive ErrBufferFull")
 	}
 	transport.Wait()
-	if transport.perMinCounter != 1 {
+	if transport.perMinCounter != 0 {
 		t.Error("shouldSend check failed")
 	}
 }
 
+func TestAsyncTransportSendRecover(t *testing.T) {
+	transport := NewAsyncTransport("", "", 1)
+	transport.SetLogger(&SilentClientLogger{})
+
+	transport.Close()
+	result := transport.Send(nil)
+	if result == nil {
+		t.Error("Expected to receive ErrChannelClosed")
+	}
+	transport.Wait()
+}
+
 func TestAsyncTransportClose(t *testing.T) {
 	transport := NewAsyncTransport("", "", 1)
 	transport.SetLogger(&SilentClientLogger{})
diff --git a/rollbar_test.go b/rollbar_test.go
index 83bf58f..3a2e367 100644
--- a/rollbar_test.go
+++ b/rollbar_test.go
@@ -133,6 +133,8 @@ func TestEverythingGeneric(t *testing.T) {
 		"hello": "rollbar",
 	})
 
+	SetItemsPerMinute(2000)
+	SetRetryAttempts(123)
 	SetLogger(&SilentClientLogger{})
 	Info(someNonstandardTypeForLogFailing{}, "I am a string and I did not fail")
 	SetLogger(nil)