<commit_before>(*
This file is part of Arakoon, a distributed key-value store. Copyright
(C) 2010 Incubaid BVBA

Licensees holding a valid Incubaid license may use this file in
accordance with Incubaid's Arakoon commercial license agreement. For
more information on how to enter into this agreement, please contact
Incubaid (contact details can be found on www.arakoon.org/licensing).

Alternatively, this file may be redistributed and/or modified under
the terms of the GNU Affero General Public License version 3, as
published by the Free Software Foundation. Under this license, this
file is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE.

See the GNU Affero General Public License for more details.
You should have received a copy of the
GNU Affero General Public License along with this program (file "COPYING").
If not, see <http://www.gnu.org/licenses/>.
*)

open Lwt
type store_maker= ?read_only:bool -> string -> Store.store Lwt.t


let collapse_until (tlog_coll:Tlogcollection.tlog_collection) 
    ((create_store:store_maker), copy_store, head_location) 
    (too_far_i:Sn.t) 
    (cb: Sn.t -> unit Lwt.t) =
  
  let new_location = head_location ^ ".clone" in
  Lwt_log.debug_f "Creating db clone at %s" new_location >>= fun () ->
  Lwt.catch (
    fun () ->
      copy_store head_location new_location true 
    ) (
     function 
      | Not_found -> Lwt_log.debug_f "head db at '%s' does not exist" head_location
      | e -> fail e
    )
  >>= fun () ->
  Lwt_log.debug_f "Creating store at %s" new_location >>= fun () ->
  create_store new_location >>= fun new_store ->
  Lwt.finalize (
    fun () ->
	  tlog_coll # get_infimum_i () >>= fun min_i ->
      let first_tlog = (Sn.to_int min_i) /  !Tlogcommon.tlogEntriesPerFile in 
	  new_store # consensus_i () >>= fun store_i ->
	  let tfs = Sn.string_of too_far_i in
	  let tlog_entries_per_file = Sn.of_int (!Tlogcommon.tlogEntriesPerFile) in
	  let processed = ref 0 in
	  let start_i = 
		begin
		  match store_i with
		    | None -> Sn.start
		    | Some i -> Sn.succ i
		end
	  in
	  
	  begin  
		if start_i >= too_far_i 
		then
		  let msg = Printf.sprintf 
			"Store counter (%s) is ahead of end point (%s). Aborting"
			(Sn.string_of start_i) tfs in
		  Lwt.fail (Failure msg) 
		else
		  let acc = ref None in
		  let maybe_log =
		    begin
		      let lo = Sn.add start_i   (Sn.of_int 10) in
		      let hi = Sn.sub too_far_i (Sn.of_int 10) in
		      function 
		        | b when b < lo || b > hi ->  Lwt_log.debug_f "%s => store" (Sn.string_of b)
		        | b when b = lo -> Lwt_log.debug " ... => store"
		        | _ -> Lwt.return ()
		    end 
		  in
		  let add_to_store (i,update) = 
		    begin 
		      match !acc with
		        | None ->
		          begin
		            let () = acc := Some(i,update) in
		            Lwt_log.debug_f "update %s has no previous" (Sn.string_of i) 
		            >>= fun () ->
		            Lwt.return ()
		          end
		        | Some (pi,pu) ->
		          if pi < i then
		            begin
		              maybe_log pi >>= fun () ->
		              begin  
		                if Sn.rem pi tlog_entries_per_file = 0L 
		                then
		                  
		                  cb (Sn.of_int (first_tlog + !processed)) >>= fun () ->
		                Lwt.return( processed := !processed + 1 )
		                else
		                  Lwt.return ()
		              end 
		              >>= fun () ->
		              Store.safe_insert_update new_store pi pu >>= fun _ ->
		              let () = acc := Some(i,update) in
		              Lwt.return ()
		            end
		          else
		            begin
		              Lwt_log.debug_f "%s => skip" (Sn.string_of pi) >>= fun () ->
		              let () = acc := Some(i,update) in
		              Lwt.return ()
		            end
		    end
		  in
		  Lwt_log.debug_f "collapse_until: start_i=%s" (Sn.string_of start_i) 
		              >>= fun () ->
		  tlog_coll # iterate start_i too_far_i add_to_store 
		              >>= fun () ->
		  new_store # consensus_i () >>= fun m_si ->
		  let si = 
		    begin
		      match m_si with
		        | None -> Sn.start
		        | Some i -> i
		    end
		  in
		  
		  Lwt_log.debug_f "Done replaying to head (%s : %s)" (Sn.string_of si) (Sn.string_of too_far_i) >>= fun() ->
		  begin
		    if si = Sn.pred (Sn.pred too_far_i) then
		      Lwt.return ()
		    else
		      let msg = Printf.sprintf "Head db has invalid counter: %s" (Sn.string_of si) in 
		      Lwt_log.debug msg >>= fun () ->
		      new_store # close () >>= fun () ->
		      Lwt.fail (Failure msg)
		  end
		    
	  end
	  >>= fun () ->
	  Lwt_log.debug_f "Relocating store to %s" head_location >>= fun () ->
	  new_store # relocate head_location
  ) 
    ( 
      fun () -> new_store # close ()
    )

let collapse_many tlog_coll 
    store_fs 
    tlogs_to_keep cb' cb =
  
  Lwt_log.debug_f "collapse_many" >>= fun () ->
  tlog_coll # get_tlog_count () >>= fun total_tlogs ->
  Lwt_log.debug_f "total_tlogs = %i; tlogs_to_keep=%i" total_tlogs tlogs_to_keep >>= fun () ->
  let ((create_store:store_maker),_,(head_location:string)) = store_fs in
  let read_only = true in
  create_store ~read_only head_location >>= fun head ->
  head # consensus_i () >>= fun head_io ->
  tlog_coll # get_last_i () >>= fun last_i ->
  Lwt_log.debug_f "head @ %s : last_i %s " (Log_extra.option2s Sn.string_of head_io) (Sn.string_of last_i)
  >>= fun () ->
  head # close () >>= fun () ->
  let head_i = match head_io with None -> Sn.start | Some i -> i in
  let lag = Sn.to_int (Sn.sub last_i head_i) in
  let npt = !Tlogcommon.tlogEntriesPerFile in
  let tlog_lag = lag / npt in
  let tlogs_to_collapse = tlog_lag - tlogs_to_keep - 1 in
  if tlogs_to_collapse <= 0
  then
    Lwt_log.info_f "Nothing to collapse..." >>= fun () -> 
    cb' 0<commit_msg>bugfix in collapser: head can also be absent<commit_after>(*
This file is part of Arakoon, a distributed key-value store. Copyright
(C) 2010 Incubaid BVBA

Licensees holding a valid Incubaid license may use this file in
accordance with Incubaid's Arakoon commercial license agreement. For
more information on how to enter into this agreement, please contact
Incubaid (contact details can be found on www.arakoon.org/licensing).

Alternatively, this file may be redistributed and/or modified under
the terms of the GNU Affero General Public License version 3, as
published by the Free Software Foundation. Under this license, this
file is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE.

See the GNU Affero General Public License for more details.
You should have received a copy of the
GNU Affero General Public License along with this program (file "COPYING").
If not, see <http://www.gnu.org/licenses/>.
*)

open Lwt
type store_maker= ?read_only:bool -> string -> Store.store Lwt.t


let collapse_until (tlog_coll:Tlogcollection.tlog_collection) 
    ((create_store:store_maker), copy_store, head_location) 
    (too_far_i:Sn.t) 
    (cb: Sn.t -> unit Lwt.t) =
  
  let new_location = head_location ^ ".clone" in
  Lwt_log.debug_f "Creating db clone at %s" new_location >>= fun () ->
  Lwt.catch (
    fun () ->
      copy_store head_location new_location true 
    ) (
     function 
      | Not_found -> Lwt_log.debug_f "head db at '%s' does not exist" head_location
      | e -> fail e
    )
  >>= fun () ->
  Lwt_log.debug_f "Creating store at %s" new_location >>= fun () ->
  create_store new_location >>= fun new_store ->
  Lwt.finalize (
    fun () ->
	  tlog_coll # get_infimum_i () >>= fun min_i ->
      let first_tlog = (Sn.to_int min_i) /  !Tlogcommon.tlogEntriesPerFile in 
	  new_store # consensus_i () >>= fun store_i ->
	  let tfs = Sn.string_of too_far_i in
	  let tlog_entries_per_file = Sn.of_int (!Tlogcommon.tlogEntriesPerFile) in
	  let processed = ref 0 in
	  let start_i = 
		begin
		  match store_i with
		    | None -> Sn.start
		    | Some i -> Sn.succ i
		end
	  in
	  
	  begin  
		if start_i >= too_far_i 
		then
		  let msg = Printf.sprintf 
			"Store counter (%s) is ahead of end point (%s). Aborting"
			(Sn.string_of start_i) tfs in
		  Lwt.fail (Failure msg) 
		else
		  let acc = ref None in
		  let maybe_log =
		    begin
		      let lo = Sn.add start_i   (Sn.of_int 10) in
		      let hi = Sn.sub too_far_i (Sn.of_int 10) in
		      function 
		        | b when b < lo || b > hi ->  Lwt_log.debug_f "%s => store" (Sn.string_of b)
		        | b when b = lo -> Lwt_log.debug " ... => store"
		        | _ -> Lwt.return ()
		    end 
		  in
		  let add_to_store (i,update) = 
		    begin 
		      match !acc with
		        | None ->
		          begin
		            let () = acc := Some(i,update) in
		            Lwt_log.debug_f "update %s has no previous" (Sn.string_of i) 
		            >>= fun () ->
		            Lwt.return ()
		          end
		        | Some (pi,pu) ->
		          if pi < i then
		            begin
		              maybe_log pi >>= fun () ->
		              begin  
		                if Sn.rem pi tlog_entries_per_file = 0L 
		                then
		                  
		                  cb (Sn.of_int (first_tlog + !processed)) >>= fun () ->
		                Lwt.return( processed := !processed + 1 )
		                else
		                  Lwt.return ()
		              end 
		              >>= fun () ->
		              Store.safe_insert_update new_store pi pu >>= fun _ ->
		              let () = acc := Some(i,update) in
		              Lwt.return ()
		            end
		          else
		            begin
		              Lwt_log.debug_f "%s => skip" (Sn.string_of pi) >>= fun () ->
		              let () = acc := Some(i,update) in
		              Lwt.return ()
		            end
		    end
		  in
		  Lwt_log.debug_f "collapse_until: start_i=%s" (Sn.string_of start_i) 
		              >>= fun () ->
		  tlog_coll # iterate start_i too_far_i add_to_store 
		              >>= fun () ->
		  new_store # consensus_i () >>= fun m_si ->
		  let si = 
		    begin
		      match m_si with
		        | None -> Sn.start
		        | Some i -> i
		    end
		  in
		  
		  Lwt_log.debug_f "Done replaying to head (%s : %s)" (Sn.string_of si) (Sn.string_of too_far_i) >>= fun() ->
		  begin
		    if si = Sn.pred (Sn.pred too_far_i) then
		      Lwt.return ()
		    else
		      let msg = Printf.sprintf "Head db has invalid counter: %s" (Sn.string_of si) in 
		      Lwt_log.debug msg >>= fun () ->
		      new_store # close () >>= fun () ->
		      Lwt.fail (Failure msg)
		  end
		    
	  end
	  >>= fun () ->
	  Lwt_log.debug_f "Relocating store to %s" head_location >>= fun () ->
	  new_store # relocate head_location
  ) 
    ( 
      fun () -> new_store # close ()
    )

let _head_i (create_store:store_maker) head_location =
  Lwt.catch
    (fun () ->
      let read_only=true in
      create_store ~read_only head_location >>= fun head ->
      head # consensus_i () >>= fun head_io ->
      head # close () >>= fun () ->
      Lwt.return head_io
    )
    (fun exn -> 
      Lwt_log.info ~exn "returning assuming no I" >>= fun () ->
      Lwt.return None
    )

let collapse_many tlog_coll 
    store_fs 
    tlogs_to_keep cb' cb =
  
  Lwt_log.debug_f "collapse_many" >>= fun () ->
  tlog_coll # get_tlog_count () >>= fun total_tlogs ->
  Lwt_log.debug_f "total_tlogs = %i; tlogs_to_keep=%i" total_tlogs tlogs_to_keep >>= fun () ->
  let ((create_store:store_maker),_,(head_location:string)) = store_fs in
  _head_i create_store head_location >>= fun head_io ->
  tlog_coll # get_last_i () >>= fun last_i ->
  Lwt_log.debug_f "head @ %s : last_i %s " (Log_extra.option2s Sn.string_of head_io) (Sn.string_of last_i)
  >>= fun () ->
  let head_i = match head_io with None -> Sn.start | Some i -> i in
  let lag = Sn.to_int (Sn.sub last_i head_i) in
  let npt = !Tlogcommon.tlogEntriesPerFile in
  let tlog_lag = lag / npt in
  let tlogs_to_collapse = tlog_lag - tlogs_to_keep - 1 in
  if tlogs_to_collapse <= 0
  then
    Lwt_log.info_f "Nothing to collapse..." >>= fun () -> 
    cb' 0
{'fraction_non_alphanumeric': 0.1071837669775814, 'fraction_numerical': 0.0026182294223531335, 'mean_word_length': 2.0413038069171434, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 235, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4972729', 'n_tokens_mistral': 4474, 'n_tokens_neox': 4018, 'n_words': 1213}