diff --git a/src/Orleans.Streaming.Grains.Tests/Streams/Scenarios/OneToMany.cs b/src/Orleans.Streaming.Grains.Tests/Streams/Scenarios/OneToMany.cs
index 6a66e4a..ee7ceb3 100644
--- a/src/Orleans.Streaming.Grains.Tests/Streams/Scenarios/OneToMany.cs
+++ b/src/Orleans.Streaming.Grains.Tests/Streams/Scenarios/OneToMany.cs
@@ -248,12 +248,14 @@ public void It_Should_Deliver_Expected_Data()
 
         public class When_Sending_Broadcast_Message_One_To_Many_Error : BaseOneToManyTest
         {
-            protected Stopwatch timer;
+            protected TimeSpan wait = TimeSpan.FromSeconds(10);
 
             protected string resultText;
+            protected Stopwatch timerText;
             protected string expectedText = "text";
 
             protected byte[] resultData;
+            protected Stopwatch timerData;
             protected byte[] expectedData = new byte[1024];
 
             public override void Prepare()
@@ -263,7 +265,9 @@ public override void Prepare()
                 Processor!.Setup(x => x.Process(It.IsAny<string>()))
                           .Callback<string>(x =>
                           {
-                              if (timer.Elapsed < TimeSpan.FromSeconds(4))
+                              timerText = timerText ?? Stopwatch.StartNew();
+
+                              if (timerText.Elapsed < wait)
                               {
                                   throw new Exception();
                               }
@@ -276,7 +280,9 @@ public override void Prepare()
                 Processor!.Setup(x => x.Process(It.IsAny<byte[]>()))
                           .Callback<byte[]>(x =>
                           {
-                              if (timer.Elapsed < TimeSpan.FromSeconds(4))
+                              timerData = timerData ?? Stopwatch.StartNew();
+
+                              if (timerData.Elapsed < wait)
                               {
                                   throw new Exception();
                               }
@@ -298,8 +304,6 @@ public override async Task Act()
             {
                 var grain = Subject.GetGrain<IEmitterGrain>(Guid.NewGuid());
 
-                timer = Stopwatch.StartNew();
-
                 await grain.BroadcastAsync(expectedText, expectedData);
             }
 
diff --git a/src/Orleans.Streaming.Grains/Grains/TransactionGrain.cs b/src/Orleans.Streaming.Grains/Grains/TransactionGrain.cs
index 345de9a..d5dbb36 100644
--- a/src/Orleans.Streaming.Grains/Grains/TransactionGrain.cs
+++ b/src/Orleans.Streaming.Grains/Grains/TransactionGrain.cs
@@ -64,6 +64,10 @@ public async Task CompleteAsync(Guid id, bool success)
                     await _subscriptions.Notify(x => x.CompletedAsync(id, success, this.GetPrimaryKeyString()));
                 }
             }
+            else
+            {
+                throw new InvalidOperationException($"No transaction with id {id} pending.");
+            }
         }
 
         public async Task<Guid?> PopAsync()