anchor.diff:      
       // until we need to stop processing, keep fetching new records when we've finished processing the previous batch
       let fetchSub        = batchProcessedEvent.Publish
  +                            .Zip(checkpointEvent.Publish, fun batchProcessedRes _ -> batchProcessedRes)
                               .TakeUntil(stopProcessing)
                               .Subscribe(fun (_, iterator) -> fetchNextRecords iterator |> Async.StartImmediate)
   
  
positive.code:                             streamName,
                              workerId,
                              shardId,
                              action     : Record -> unit) =
      let batchReceivedEvent          = new Event<Iterator * Record seq>()
      let recordProcessedEvent        = new Event<Record>()
      let processErroredEvent         = new Event<Record * Exception>()
      let batchProcessedEvent         = new Event<int * Iterator>()
  
      let heartbeatEvent              = new Event<unit>()
      let conditionalCheckFailedEvent = new Event<unit>()
  
      let catchCondCheck f = 
          try
              f()
          with
          | :? ConditionalCheckFailedException -> 
              conditionalCheckFailedEvent.Trigger()
              reraise()
  
      let updateHeartbeat _ = 
          (fun () -> DynamoDBUtils.updateHeartbeat dynamoDB tableName workerId shardId |> Async.RunSynchronously)
          |> catchCondCheck
  
      let updateCheckpoint seqNum =
          (fun () -> DynamoDBUtils.updateCheckpoint seqNum dynamoDB tableName workerId shardId |> Async.RunSynchronously)
          |> catchCondCheck
  
      let fetchNextRecords iterator = 
          async {
  
negative.code: open System
  open System.Reactive
  open System.Reactive.Linq
  open System.Threading
  
  open Amazon
  open Amazon.DynamoDBv2
  