Checksums

doc/design/rolling-checksums.rst

@@ -0,0 +1,93 @@
+=================
+Rolling Checksums
+=================
+
+Problem
+=======
+If a node crashed, and failed to write some tlog entries to disk, this is not detected by Arakoon. The node announces it's in sync up to the last entry in the tlogs, even if other nodes diverged while the node was offline.
+
+Example
+-------
+Consider this situation:
+
++----------------------------------+------------------------------------+------------------------------------+
+| node0                            | node1                              | node 2                             |
++==================================+====================================+====================================+
+| 0:(Vm (node0,0.000000))          | 0:(Vm (node0,0.000000))            | 0:(Vm (node0,0.000000))            |
++----------------------------------+------------------------------------+------------------------------------+
+| 1:(Vc ([Set;"a";1;"...";],false) | 1:(Vc ([Set;"a";1;"...";],false)   | 1:(Vc ([Set;"a";1;"...";],false)   |
++----------------------------------+------------------------------------+------------------------------------+
+| 2:(Vc ([Set;"b";1;"...";],false) | *2:(Vc ([Set;"b";1;"...";],false)* | *2:(Vc ([Set;"b";1;"...";],false)* |
++----------------------------------+------------------------------------+------------------------------------+
+| 3:(Vc ([Set;"c";1;"...";],false) | *3:(Vc ([Set;"c";1;"...";],false)* |                                    |
++----------------------------------+------------------------------------+------------------------------------+
+
+Node1 and node2 crashed, and the last tlog entries were lost. They are restarted, while node0 is still offline. When node0 comes back, this will result in the following situation:
+
++--------------------------------------+----------------------------------+----------------------------------+
+| node0                                | node1                            | node 2                           |
++======================================+==================================+==================================+
+| 0:(Vm (node0,0.000000))              | 0:(Vm (node0,0.000000))          | 0:(Vm (node0,0.000000))          |
++--------------------------------------+----------------------------------+----------------------------------+
+| 1:(Vc ([Set;"a";1;"...";],false)     | 1:(Vc ([Set;"a";1;"...";],false) | 1:(Vc ([Set;"a";1;"...";],false) |
++--------------------------------------+----------------------------------+----------------------------------+
+| **2:(Vc ([Set;"b";1;"...";],false)** | **2:(Vm (node1,0.000000))**      | **2:(Vm (node1,0.000000))**      |
++--------------------------------------+----------------------------------+----------------------------------+
+| 3:(Vc ([Set;"d";1;"...";],false)     | 3:(Vc ([Set;"d";1;"...";],false) | 3:(Vc ([Set;"d";1;"...";],false) |
++--------------------------------------+----------------------------------+----------------------------------+
+
+Checksums
+=========
+This problem can be solved by using a rolling checksum, computed over all the entries in the tlogs. This checksum should be the same for all nodes. The checksum is part of the value that is synced with multi-paxos.
+
+1. The client sends a request to the master node.
+2. The master computes the rolling checksum, and makes a value of this checksum and the update commands.
+3. This value is sent to the slaves in an accept request.
+4. The slaves compute the rolling checksum, and compare it with the checksum in the value.
+5. If the checksums are equal, the tlogs are in sync, the value is written to the tlogs, and the algorithm proceeds as usual.
+6. If the checksums are different, something bad happened. The node halts, and the tlogs need to be inspected manually.
+
+Remark
+------
+When several consecutive entries in the tlogs have the same number, it is only the last one that is agreed upon by multi-paxos, and thus only this entry is used in the computation of the checksum.
+
+Tlog Specification
+==================
+* Serial number (int64)
+* Crc-32 checksum of Cmd (int32)
+* Cmd
+  - Value
+  - Marker, optional (string option)
+
+Older value format
+------------------
+* Update
+  - Update type (int32 between 1 and 16)
+  - Update details (depends on type)
+* Synced (bool)
+
+Old value format
+----------------
+* 0xff (int32)
+* Value type (char 'c' or 'm')
+* Value details (depends on type)
+
+New value format
+----------------
+* 0x100 (int32)
+* Checksum (int32 if crc-32 is used)
+* Value type (char 'c' or 'm')
+* Value details (depends on type)
+
+Checksums in store
+==================
+The current tlog index and checksum are stored in the local store. When a node starts, they are compared with the values in the tlog.
+Thus, after a collapse, the checksum of the last collapsed value is saved in the head database. Therefore the rolling tlog checksum is a continuation of the checksum in the head database.
+
+Upgrade Path
+============
+To upgrade Arakoon to the new version, with a new tlog format, the nodes need to be restarted. A node that restarts after the upgrade can still read the old tlogs, and the checksums of these values will be set to None. Values with checksum None will never be valuated.
+
+Nodes that need to do a catchup will do this as usual. The first received values will have checksum None, and are written to the tlogs in the old format, until all values from before the upgrade are synced. All values that are created after the upgrade will get a checksum. The checksum of the first value is a normal checksum (not depending on previous values), and all following checksums are rolling.
+
+New and old nodes can not and will not communicate. If the nodes are restarted one by one, the old nodes will keep going as long as possible, while the new nodes can't make progress because they don't have a majority. When the critical point is reached, the new nodes will do a catchup and take over.

src/client/arakoon_client.mli

@@ -85,7 +85,7 @@ class type client = object
 
   method nop : unit -> unit Lwt.t
   (** [nop ()] is a paxos no-operation.
-   *)
+  *)
   method confirm: key -> value -> unit Lwt.t
   (** [confirm key value] does nothing if this value was already
       associated to the key, otherwise, it behaves as [set key value]
@@ -118,7 +118,7 @@ class type client = object
       [replace key wanted] assigns the wanted value to the key,
       and returns the previous assignment (if any) for that key.
       If wanted is None, the binding is deleted.
-   *)
+  *)
 
   method ping: string -> string -> string Lwt.t
 

src/client/remote_nodestream.ml

@@ -25,9 +25,9 @@ let section = Logger.Section.main
 
 class type nodestream = object
   method iterate:
-    Sn.t -> (Sn.t * Value.t -> unit Lwt.t) ->
-    Tlogcollection.tlog_collection ->
-    head_saved_cb:(string -> unit Lwt.t) -> unit Lwt.t
+    Sn.t -> f_entry:(Sn.t * Value.t -> unit Lwt.t) ->
+    f_head:(Lwt_io.input_channel -> unit Lwt.t) ->
+    f_file:(string -> int64 -> Lwt_io.input_channel -> unit Lwt.t) -> unit Lwt.t
 
   method collapse: int -> unit Lwt.t
 
@@ -50,18 +50,13 @@ class type nodestream = object
   method drop_master: unit -> unit Lwt.t
 end
 
-class remote_nodestream ((ic,oc) as conn) = 
-        (object
-  method iterate (i:Sn.t) (f: Sn.t * Value.t -> unit Lwt.t)  
-    (tlog_coll: Tlogcollection.tlog_collection) 
-    ~head_saved_cb
-    =
+class remote_nodestream ((ic,oc) as conn) = (object
+  method iterate i ~f_entry ~f_head ~f_file =
     let outgoing buf =
       command_to buf LAST_ENTRIES2;
       Sn.sn_to buf i
     in
     let incoming ic =
-      let save_head () = tlog_coll # save_head ic in
       let last_seen = ref None in
       let rec loop_entries () =
         Sn.input_sn ic >>= fun i2 ->
@@ -78,7 +73,7 @@ class remote_nodestream ((ic,oc) as conn) =
               Llio.input_int32 ic >>= fun _chksum ->
               Llio.input_string ic >>= fun entry ->
               let value = Value.value_from (Llio.make_buffer entry 0) in
-              f (i2, value) >>= fun () ->
+              f_entry (i2, value) >>= fun () ->
               loop_entries ()
             end
         end
@@ -94,10 +89,8 @@ class remote_nodestream ((ic,oc) as conn) =
           end
         | 2 ->
           begin
-            Logger.info_f_ "save_head" >>= fun ()->
-            save_head () >>= fun () ->
-            let hf_name = tlog_coll # get_head_name () in
-            head_saved_cb hf_name >>= fun () ->
+            Logger.info_f_ "save_head" >>= fun () ->
+            f_head ic >>= fun () ->
             loop_parts ()
           end
         | 3 ->
@@ -106,7 +99,7 @@ class remote_nodestream ((ic,oc) as conn) =
             Llio.input_string ic >>= fun name ->
             Llio.input_int64 ic >>= fun length ->
             Logger.info_f_ "got %s (%Li bytes)" name length >>= fun () ->
-            tlog_coll # save_tlog_file name length ic >>= fun () ->
+            f_file name length ic >>= fun () ->
             loop_parts ()
           end
         | x  -> Llio.lwt_failfmt "don't know what %i means" x

src/client/remote_nodestream.mli

@@ -21,9 +21,9 @@ open Ncfg
 
 class type nodestream = object
   method iterate:
-    Sn.t -> (Sn.t * Value.t -> unit Lwt.t) ->
-    Tlogcollection.tlog_collection ->
-    head_saved_cb:(string -> unit Lwt.t) -> unit Lwt.t
+    Sn.t -> f_entry:(Sn.t * Value.t -> unit Lwt.t) ->
+    f_head:(Lwt_io.input_channel -> unit Lwt.t) ->
+    f_file:(string -> int64 -> Lwt_io.input_channel -> unit Lwt.t) -> unit Lwt.t
 
   method collapse: int -> unit Lwt.t
 

src/main/replay_main.ml

@@ -49,7 +49,7 @@ let replay_tlogs tlog_dir tlf_dir db_name end_i =
              | None ->
                begin
                  Tlc2.get_last_tlog tlog_dir tlf_dir >>= fun (_new_c,fn) ->
-                 Tlc2._validate_one fn "" ~check_marker:false >>= fun (last, _index) ->
+                 Tlc2._validate_one fn "" ~check_marker:false >>= fun (last, _, _index) ->
                  let i =
                    match last with
                      | None -> Sn.start

src/msg/tcp_messaging.ml

@@ -69,7 +69,7 @@ class tcp_messaging
   my_addresses my_cookie (drop_it: drop_function)
   max_buffer_size ~stop =
 
-  let _MAGIC = 0xB0BAFE7L in
+  let _MAGIC = 0x53E7965CL in
   let _VERSION = 1 in
   let my_ips, my_port = my_addresses in
   let my_ip = List.hd my_ips in

src/node/catchup.ml

@@ -24,6 +24,7 @@ open Log_extra
 open Tlogcommon
 
 exception StoreAheadOfTlogs of (Int64.t * Sn.t)
+exception StoreChecksumError of (Sn.t * Checksum.Crc32.t option * Checksum.Crc32.t option)
 exception StoreCounterTooLow of string
 
 let with_connection ~tls_ctx address f = match tls_ctx with
@@ -86,9 +87,11 @@ let head_saved_epilogue hfn tlog_coll =
   S.make_store ~lcnum:default_lcnum
     ~ncnum:default_ncnum ~read_only:true hfn >>= fun store ->
   let hio = S.consensus_i store in
+  let hcso = S.get_checksum store in
   S.close store ~flush:false ~sync:false >>= fun () ->
   Logger.info_ "closed head" >>= fun () ->
   begin
+    tlog_coll # set_previous_checksum hcso;
     match hio with
       | None -> Lwt.return ()
       | Some head_i ->
@@ -98,14 +101,16 @@ let head_saved_epilogue hfn tlog_coll =
         end
   end
 
+
 let stop_fuse stop =
   if !stop
   then
     Lwt.fail Canceled
   else
     Lwt.return ()
 
-let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id  mr_name ((module S : Store.STORE with type t = s),store,tlog_coll)
+let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id mr_name
+      ((module S : Store.STORE with type t = s), store, (tlog_coll:Tlogcollection.tlog_collection))
   =
   let current_i = tlog_coll # get_last_i () in
   Logger.info_f_ "catchup_tlog %s" (Sn.string_of current_i) >>= fun () ->
@@ -114,8 +119,26 @@ let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id  mr_name ((mo
   let mr_addresses = Node_cfg.client_addresses mr_cfg
   and mr_name = Node_cfg.node_name mr_cfg in
   Logger.info_f_ "getting last_entries from %s" mr_name >>= fun () ->
-  let head_saved_cb hfn =
-    Logger.info_f_ "head_saved_cb %s" hfn >>= fun () ->
+
+  let r_validate = ref false in
+  let validate () =
+    let _validate = !r_validate in
+    begin
+      r_validate := false;
+      _validate
+    end
+  in
+
+  let f_entry (i, value) =
+    let validate = validate () in
+    tlog_coll # log_value_explicit i value ~validate:validate false None >>= fun _ ->
+    stop_fuse stop
+  in
+
+  let f_head ic =
+    tlog_coll # save_head ic >>= fun () ->
+    let hfn = tlog_coll # get_head_name () in
+    Logger.info_f_ "head_saved %s" hfn >>= fun () ->
     head_saved_epilogue hfn tlog_coll >>= fun () ->
     let when_closed () =
       Logger.debug_ "when_closed" >>= fun () ->
@@ -126,20 +149,20 @@ let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id  mr_name ((mo
     Lwt.return ()
   in
 
+  let f_file name length ic =
+    let validate = validate () in
+    tlog_coll # save_tlog_file ~validate:validate name length ic
+  in
+
   let copy_tlog connection =
     make_remote_nodestream cluster_id connection >>= fun (client:nodestream) ->
-    let f (i,value) =
-      tlog_coll # log_value_explicit i value false None >>= fun _ ->
-      stop_fuse stop
-    in
-
-    client # iterate current_i f tlog_coll ~head_saved_cb
+    client # iterate current_i ~f_entry ~f_head ~f_file
   in
 
   Lwt.catch
     (fun () ->
-      _with_client_connection  ~tls_ctx mr_addresses copy_tlog >>= fun () ->
-      Logger.info_f_ "catchup_tlog completed"
+       _with_client_connection  ~tls_ctx mr_addresses copy_tlog >>= fun () ->
+       Logger.info_f_ "catchup_tlog completed"
     )
     (fun exn -> Logger.warning_ ~exn "catchup_tlog failed")
   >>= fun () ->
@@ -297,9 +320,28 @@ let verify_n_catchup_store (type s) ~stop me ?(apply_last_tlog_value=false) ((mo
     (Sn.string_of too_far_i) (Sn.string_of current_i) (io_s si_o) >>= fun () ->
    match too_far_i, si_o with
     | i, None when i <= 0L -> Lwt.return ()
-    | i, Some j when i = j -> Lwt.return ()
+    | i, Some j when i = j ->
+      begin
+        let store_cs = S.get_checksum store in
+        let tlog_cs = tlog_coll # get_previous_checksum (Sn.succ i) in
+        if store_cs <> tlog_cs
+        then Lwt.fail (StoreChecksumError (i, store_cs, tlog_cs))
+        else Lwt.return ()
+      end
     | i, Some j when i > j ->
-      catchup_store ~stop me ((module S),store,tlog_coll) too_far_i
+      let entry = ref None in
+      let check e = Lwt.return (entry := Some e) in
+      tlog_coll # iterate j (Sn.succ j) check >>= fun () ->
+      let store_cs = S.get_checksum store in
+      let tlog_cs =
+        match !entry with
+          | None -> None
+          | Some e -> Value.checksum_of (Entry.v_of e) in
+      if store_cs <> tlog_cs
+      then
+        Lwt.fail (StoreChecksumError (j, store_cs, tlog_cs))
+      else
+        catchup_store ~stop me ((module S),store,tlog_coll) too_far_i
     | _, None ->
       catchup_store ~stop me ((module S),store,tlog_coll) too_far_i
     | _,_ ->

src/node/catchup_test.ml

@@ -38,8 +38,9 @@ let _fill tlog_coll n =
         let k = Printf.sprintf "key%i" i
         and v = Printf.sprintf "value%i" i in
         let u = Update.Set (k,v) in
-        let value = Value.create_client_value [u] sync in
-        tlog_coll # log_value (Sn.of_int i) value >>= fun () ->
+        let sni = Sn.of_int i in
+        let value = Value.create_client_value tlog_coll sni [u] sync in
+        tlog_coll # log_value sni value >>= fun () ->
         _loop (i+1)
       end
   in
@@ -58,11 +59,11 @@ let _fill2 tlog_coll n =
         in
         let u = Update.Set(k,v) in
         let u2 = Update.Set(k2,v2) in
-        let value = Value.create_client_value [u] sync in
-        let value2 = Value.create_client_value [u2] sync in
         let sni = Sn.of_int i in
-        tlog_coll # log_value  sni value  >>= fun () ->
-        tlog_coll # log_value  sni value2 >>= fun () ->
+        let value = Value.create_client_value tlog_coll sni [u] sync in
+        let value2 = Value.create_client_value tlog_coll sni [u2] sync in
+        tlog_coll # log_value sni value  >>= fun () ->
+        tlog_coll # log_value sni value2 >>= fun () ->
         _loop (i+1)
       end
   in
@@ -84,9 +85,9 @@ let _fill3 tlog_coll n =
         let u = Update.Set(k,v) in
         let u2 = Update.Set(k2,v2) in
         let u3 = Update.Sequence [Update.Set(k3,v3); Update.Assert_exists("nonExistingKey")] in
-        let value = Value.create_client_value [u; u3] sync in
-        let value2 = Value.create_client_value [u2; u3] sync in
         let sni = Sn.of_int i in
+        let value = Value.create_client_value tlog_coll sni [u; u3] sync in
+        let value2 = Value.create_client_value tlog_coll sni [u2; u3] sync in
         tlog_coll # log_value sni value  >>= fun () ->
         tlog_coll # log_value sni value2 >>= fun () ->
         _loop (i+1)