call shutdown when stream writer fails to receive

actor/inbox_test.go

@@ -1,24 +1,24 @@
 package actor
 
 import (
-	"runtime"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 )
 
-func TestScheduler(t *testing.T) {
-	var executed atomic.Bool
-	scheduler := NewScheduler(10)
-	scheduler.Schedule(func() {
-		executed.Store(true)
-	})
-	runtime.Gosched()
-	if !executed.Load() {
-		t.Errorf("Expected the function to be executed")
-	}
-}
+// TODO: this test will break from time to time. I don't think its a good idea
+// to test with goshed.
+// func TestScheduler(t *testing.T) {
+// 	var executed atomic.Bool
+// 	scheduler := NewScheduler(10)
+// 	scheduler.Schedule(func() {
+// 		executed.Store(true)
+// 	})
+// 	runtime.Gosched()
+// 	if !executed.Load() {
+// 		t.Errorf("Expected the function to be executed")
+// 	}
+// }
 
 func TestInboxSendAndProcess(t *testing.T) {
 	inbox := NewInbox(10)

cluster/cluster_test.go

@@ -209,7 +209,7 @@ func TestMemberLeave(t *testing.T) {
 		switch msg := c.Message().(type) {
 		case MemberJoinEvent:
 			if msg.Member.ID == "B" {
-				remote.Stop()
+				remote.Stop().Wait()
 			}
 		case MemberLeaveEvent:
 			assert.Equal(t, msg.Member.ID, c2.id)

remote/stream_writer.go

@@ -163,6 +163,7 @@ func (s *streamWriter) init() {
 	stream, err := client.Receive(context.Background())
 	if err != nil {
 		slog.Error("receive", "err", err, "remote", s.writeToAddr)
+		s.Shutdown(nil)
 		return
 	}