anchor.diff:                        (folder : 'State -> 'T -> 'State) 
                         (combiner : 'State -> 'State -> 'State) 
                         (state : unit -> 'State) (stream : CloudStream<'T>) : CloudStream<'Key * 'State> = 
  -        let collectorf (totalWorkers : int) () =
  +        let collectorf (totalWorkers : int) = cloud {
               let dict = new ConcurrentDictionary<'Key, 'State ref>()
  -            { new Collector<'T,  seq<int * seq<'Key * 'State>>> with
  +            return
  +              { new Collector<'T,  seq<int * seq<'Key * 'State>>> with
                   member self.Iterator() = 
                       {   Index = ref -1; 
                           Func =
  
positive.code:     /// The number of concurrently executing tasks
      abstract DegreeOfParallelism : int option ref
      /// Applies the given collector to the CloudStream.
      abstract Apply<'S, 'R> : (unit -> Collector<'T, 'S>) -> ('S -> Cloud<'R>) -> ('R -> 'R -> 'R) -> Cloud<'R>
  
  [<RequireQualifiedAccess; CompilationRepresentation(CompilationRepresentationFlags.ModuleSuffix)>]
  /// Provides basic operations on CloudStreams.
  
negative.code:                             else
                                  let! results2 = restPartitions 
                                                  |> Set.toArray 
                                                  |> Array.map (fun pid -> cloud { let! r = createTask pid collectorf in return pid,r }) 
                                                  |> Cloud.Parallel
                                  let final = Seq.append results1 results2
                                              |> Seq.sortBy (fun (p,_) -> p)
  