Something went wrong. Try again.
My own corner of monopam
Something went wrong. Try again.
116 kB · 2653 lines
OCaml
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654(*--------------------------------------------------------------------------- Copyright (c) 2025 Thomas Gazagnaire. All rights reserved. SPDX-License-Identifier: ISC ---------------------------------------------------------------------------*)
(* Opening, closing, syncing and recovering a {!Handle.t}, and the transaction and savepoint boundaries its durability points hang from. It reads and writes every field of the record directly: the record is the shared state the rest of the engine mutates, not an abstraction over it. *)
open Modelopen Handleopen Durabilityopen Page1module Ident = Catalog.Ident
let ident_equal = Ident.equal
(* Refuse a write on a handle that may only read. One page through would reach the database at the next commit, over whatever the writer committed in the meantime, so every write door asks this first. [Invalid_argument] because it is the caller's mistake about the handle, not a state of the database. *)let check_writable t what = if t.read_only then Fmt.invalid_arg "%s: this database handle is read-only" what
(* Run [f] where a cancellation cannot reach it. A cancellation stops a fiber at its next Eio call, and [Fun.protect] is no answer: the finally runs in the still-cancelled context. Asked only of a handle with a file, since an in-memory database makes no Eio call and has no scheduler to ask. *)let protecting t f = if Option.is_some t.db_path then Eio.Cancel.protect f else f ()
(* The sqlite_master table b-tree: rooted at page 1 with its page header at byte 100, after the database header. The root page never moves (the file format fixes it), so the handle can be rebuilt on demand. *)let master_btree t = Btree.Table.open_ ~header_offset:100 t.pager ~root_page:1
let master_entries t = let entries = ref [] in let rowid = ref 1 in let add (name, tbl_name, root_page, sql, type_) = entries := (!rowid, name, tbl_name, root_page, sql, type_) :: !entries; incr rowid in List.iter (fun gt -> List.iter add (table_master_rows gt)) t.all_tables; (* Preserve views and triggers *) List.iter (fun rm -> add (rm.name, rm.tbl_name, rm.root_page, rm.sql, rm.type_)) t.extra_master; List.rev !entries
(* Lay an empty database out in [pager]: page 1 alone, carrying the 100-byte database header with the sqlite_master root at byte 100. Nothing else -- a fresh sqlite3 database has no table in it, and the key-value store's own table is created the first time something writes through {!Kv}. *)let init_pages pager = let page1 = Btree.Pager.allocate pager in let _master = Btree.Table.v ~root_page:page1 ~header_offset:100 pager in ()
(* btree.c, newDatabase: a database of no pages has page 1 laid out by the first write transaction, the write that needs it, and not by the open. *)let new_database t = if Btree.Pager.page_count t.pager = 0 then init_pages t.pager
(* The rows [entries] adds after [written], when [written] is a prefix of it. *)let rec appended written entries = match (written, entries) with | [], rest -> Some rest | w :: written, e :: entries when w = e -> appended written entries | _ :: _, _ -> None
(* Bring page 1 and the schema it roots up to the in-memory state: the sqlite_master rows go through the generic table b-tree machinery, so a schema larger than page 1 spills into interior/leaf child pages as the file format allows, and the 100-byte database header is refreshed last so it records the freelist and page count those writes leave behind. As sqlite3 writes sqlite_master only where a statement changes the schema (build.c, sqlite3NestedParse), rows already written stay where they are: rows that extend the ones last written are inserted alone, and the table is rewritten whole only when an earlier row changed. A database of no pages has its page 1 laid out first: a header write is a write transaction. *)let rebuild_page1 t = new_database t; let entries = master_entries t in let insert master rows = List.iter (fun (rowid, name, tbl_name, root_page, sql, type_) -> Btree.Table.insert master ~rowid:(Int64.of_int rowid) (master_record ~type_ ~name ~tbl_name ~root_page ~sql)) rows in let fresh = match t.master_written with | Some (pager, written) when pager == t.pager -> appended written entries | Some _ | None -> None in t.master_written <- None; let master = master_btree t in (match fresh with | Some rows -> insert master rows | None -> Btree.Table.clear master; insert master entries); t.master_written <- Some (t.pager, entries); let buf = Btree.Pager.read_mut t.pager 1 in write_db_header buf ~page_size:(Btree.Pager.page_size t.pager) ~page_count:(Btree.Pager.page_count t.pager) ~freelist_head:(Btree.Pager.freelist_head t.pager) ~freelist_count:(Btree.Pager.freelist_count t.pager) ~user_version:t.user_version ~reserved:(Btree.Pager.reserved t.pager) ~change_counter:t.change_counter (* The header states which of the two ways this database's commits are made durable, because that is what tells a reader whether a [-wal] beside it means anything: one meeting the legacy version opens in rollback mode and never looks at the log, however many committed transactions it holds. *) ~file_format: (match t.journal_mode with | `Wal -> wal_file_format | `Journal -> legacy_file_format)
(* The zone a handle keeps when the open injected none: UTC, where 'localtime' and 'utc' are the no-ops the sqlite3 CLI performs under TZ=UTC. There is no timezone database here to read a real offset out of, and guessing one would be a different answer from the CLI's rather than a better one, so this is what a caller that says nothing gets. *)let utc _ = 0
(* A clock as a handle stores it. The doors take the widest resource that carries a clock -- a mock one carries other tags too -- and the field holds the one interface it reads through. *)let stored_clock c = (c :> float Eio.Time.clock_ty Eio.Resource.t)
(* A log left to grow until close is a log every page of the database ends up in, and each of those pages has to stay cached until it reaches the file -- so reading or writing a large database would mean holding all of it. Folding the log into the file once it passes a threshold bounds both. SQLite calls this wal_autocheckpoint and defaults it to a thousand pages; so do we. *)let autocheckpoint_pages = 1000
(* SQLite's PRAGMA cache_size until one is set: 2000 KiB (sqliteLimit.h, SQLITE_DEFAULT_CACHE_SIZE). *)let default_cache_size = -2000
(* Every field a handle starts life with, whichever door made it. Only the pager, the file, what handles tell each other and how much has been read differ between the three doors; the rest is empty and cold in all three. *)let handle ~busy ~pager ~file ~writable_file ~read_only ~readers ~claim ~sw ~db_path ~synchronous ~journal_mode ~clock ~zone ~file_change_seen ~wal_seen ~wal_pages () = let lock_file = match (db_path, file) with | Some path, Some file -> Some (Writer_lock.file ~writable:(Option.is_some writable_file) path file) | _ -> None in (* A record lock of this process may go through the handle's descriptor. The switch releases in reverse order, so this, put on after the file was opened on it, takes the descriptor out of Eio before Eio's own close of it, which would let go of every record lock of the process on the file: the lock keeps it where it goes through it. *) (match (sw, lock_file) with | Some sw, Some lock -> Eio.Switch.on_release sw (fun () -> Writer_lock.close_later lock) | _ -> ()); { pager; file; writable_file; lock_file; read_only; readers; claim; physical_failed = false; detached_stamp = None; wal = None; unreliable = None; index_hold = None; shared = None; observed = false; busy; open_door = None; read_depth = 0; live_read = false; scan_door = None; sw; db_path; named_tables = []; all_tables = []; extra_master = []; master_written = None; insert_rowids = Ident.Tbl.create 8; txn = None; savepoints = []; stmt_depth = 0; violation_conflict = Sql.Ast.Unspecified; synchronous; user_version = 0; change_counter = 1; pages_written = 0; file_change_seen; wal_seen; alone_seen = None; wal_pages; wal_autocheckpoint = autocheckpoint_pages; cache_size = default_cache_size; copy_back = None; journal_mode; trigger_scope = []; last_insert_rowid = 0L; clock; zone; now = None; trigger_cache = None; fk_enforce = false; query_cache = Hashtbl.create 64; plan_cache = Hashtbl.create 64; hash_budget = Catalog.Join.budget; scanned = Catalog.Vdbe.counter (); }
(* The database file [path] names, for a handle that reads it, or with [open_out_db] writes it too: a descriptor a close of this process kept on that file with the same access before a fresh one, as os_unix.c unixOpen asks findReusableFd first.
A directory is refused at the open, as sqlite3 answers SQLITE_CANTOPEN: a read-only descriptor on one opens, and only its first read fails, with an EISDIR outside [Eio.Io]; a writable one fails to open, unclassified. *)let open_in_db ~sw path = match Writer_lock.reuse ~sw ~writable:false path with | Some file -> (file :> Eio.File.ro_ty Eio.Resource.t) | None -> let file = Eio.Path.open_in ~sw path in if (Eio.File.stat file).kind = `Directory then begin Eio.Resource.close file; Model.fail_is_a_directory path end; file
let open_out_db ~sw path = match Writer_lock.reuse ~sw ~writable:true path with | Some file -> file | None -> ( match Eio.Path.open_out ~sw ~create:`Never path with | file -> (file :> Eio.File.rw_ty Eio.Resource.t) | exception Eio.Io _ when Eio.Path.is_directory path -> Model.fail_is_a_directory path)
(* Run the rest of an open whose descriptor is [lock]'s, and give that descriptor back where the rest refuses the database: an open takes the descriptor before it can refuse the file, so without this a retry pays one per attempt. It goes back through {!Writer_lock.close_later}, as a close does: closing any descriptor on the file releases every record lock this process holds on it, among them those of the other handles. *)let opening lock f = match f () with | t -> t | exception e -> Writer_lock.close_later lock; raise e
(* Run the rest of an open under this handle's claim on the database file [file], opened by [path], and give the claim back where the rest refuses the database. A handle that may write takes the file for itself -- it allocates pages and folds a log it alone removes -- so a file another process holds is refused as SQLITE_BUSY. *)let claiming ~busy path file f = let lock = Writer_lock.file ~writable:true path (file :> Eio.File.ro_ty Eio.Resource.t) in match Writer_lock.take_session ~busy lock with | None -> Model.fail_busy "database is locked: %a is open for writing in another process" Eio.Path.pp path | Some claim -> ( match f claim with | t -> t | exception e -> Writer_lock.release claim; raise e)
(* SHARED on the database file [lock], opened by [path], waiting up to as [busy] waits for a writer of another process to let it go, as sqlite3_busy_timeout waits, and refused as SQLITE_BUSY after that. A read-only handle reads under it: a file another process writes moves under a read, its commits and checkpoints putting pages into it, and SHARED never coexists with the EXCLUSIVE that writer holds (lockingv3.html, section 3). *)let take_shared ~busy path lock = match Writer_lock.take_shared ~busy lock with | Some lock -> lock | None -> Model.fail_busy "database is locked: %a is open for writing in another process" Eio.Path.pp path
(* Run [f] under SHARED on [lock] and give it back however [f] ends. *)let sharing ~busy path lock f = let shared = take_shared ~busy path lock in Fun.protect ~finally:(fun () -> Writer_lock.release shared) f
(* A session on the WAL-index of the database at [path], whose descriptor is [file], begun by [start], waiting as [busy] waits while the index refuses it, as sqlite3's busy handler waits. The index is reached by the path, as sqlite3 reaches it, and created with the open file's mode (os_unix.c, unixOpenSharedMemory and findCreateFileMode). *)let index_session ?(on_failure = Wal_index.close) ?held ~sw ~busy path file start = let refused = ref "its WAL-index is in use" in let perm = (Eio.File.stat file).perm land 0o777 in (* An index another session is resetting is ready within moments: it is waited for as walTryBeginRead waits for a state that changes under it, before the busy handler is asked. *) let rec attach n = match Wal_index.attach ~sw ?held ~perm path with | `Busy when n < Wal_index.tries -> Wal_index.backoff n; attach (n + 1) | attached -> attached in match Writer_lock.retry ~busy (fun () -> match attach 0 with | `Unwritable | `Missing -> Some `Unwritable | `Busy -> None | `Attached index -> ( match start index with | Ok () -> Some (`Held index) | Error reason -> refused := reason; Wal_index.close index; None | exception e -> let bt = Printexc.get_raw_backtrace () in on_failure index; Printexc.raise_with_backtrace e bt)) with | Some held -> held | None -> `Busy !refused
(* What a handle read under a WAL-index session: the index's header, which every commit and every restart of the log changes, and the database file's size and version bytes (pager.c, dbFileVers), which the checkpoint that empties the log before it is removed with its index changes. An index rebuilt over no log has the header of any other, so the header alone does not tell that the file moved under the cache. *)let index_stamp ~file index = let size = Optint.Int63.to_int (Eio.File.stat file).size in let version = if size < 40 then "" else begin let buf = Cstruct.create 16 in Eio.File.pread_exact file ~file_offset:(Optint.Int63.of_int 24) [ buf ]; Cstruct.to_string buf end in Fmt.kstr Option.some "index:%s:%d:%s" (Wal_index.stamp index) size version
(* Keep the index a session of [t] reached open until [t] closes, as a sqlite3 connection keeps its index mapped, so the next session of this process reads the index as it stands rather than rebuilding it from the log. *)let hold_index ~sw t index = match t.index_hold with | Some held when Wal_index.holds held index -> () | previous -> Option.iter Wal_index.release previous; t.index_hold <- Some (Wal_index.hold ~sw index)
let drop_index_hold t = Option.iter Wal_index.release t.index_hold; t.index_hold <- None
(* Give a handle's claim back when the switch its open was made on ends and no close came, take again whatever claim of this process survives that, and hand back the way to record the handle. A read-only handle gives back the SHARED a read left it holding and reaches the second step too: closing any descriptor on the file drops every record lock this process holds, so a reader merely dropped would take a writer's claim away. The hook goes on before the open takes anything, because the switch releases in reverse order and the claim must be taken again after the descriptor close. The recorded handle says whether the steps have run. *)let claim_back_at_switch_end ~sw = let claimed = ref None in Eio.Switch.on_release sw (fun () -> match !claimed with | Some t when not (is_closed t) -> (* A handle dropped with its switch ends as a crash would: its committed work is in the log already, and what it held is given back, the checkpoint count a read left entered among it. *) if t.live_read then begin Option.iter Readers.leave t.readers; t.live_read <- false end; Option.iter Wal_index.close t.wal; t.wal <- None; Option.iter Wal_index.end_unreliable t.unreliable; t.unreliable <- None; drop_index_hold t; Option.iter Writer_lock.release t.claim; Option.iter Writer_lock.release t.shared; t.shared <- None; t.observed <- false; Option.iter Writer_lock.reassert t.lock_file | Some _ | None -> ()); fun t -> claimed := Some t; t
(* Write page 1 of a database of no pages into its file now, as sqlite3's PRAGMA journal_mode=WAL does: setting the header's version bytes is a write transaction (btree.c, sqlite3BtreeSetVersion), and that lays page 1 out ({!new_database}). A WAL database is never left at no pages, since a log beside a file of no pages is one the next sqlite3 deletes (pager.c, pagerOpenWalIfPresent). *)let lay_out_wal t = rebuild_page1 t; Btree.Pager.sync t.pager; t.file_change_seen <- t.change_counter
(* Make a new database at [path] and hand back a handle on it.
The directory above [path] is the caller's: a missing component raises [Not_found] rather than being created. The file's name is barriered with its pages, since a [synchronous=Off] handle creates no log. A log already at the path is removed rather than read: its frames name pages of a file that is not there. A journal there is the remnant of a database of no pages, which a connection holding RESERVED deletes unread (pager.c, hasHotJournal). A rollback-journal database is left at no pages until its first write ({!new_database}); a WAL one gets its page 1 now ({!lay_out_wal}). *)let init ~sw ~claimed ~busy ?(synchronous = `Full) ?(journal_mode = `Wal) ~clock ~zone path = check_sidecars ~db:None path; ignore (remove_wal (wal_path path) : removal); let file = Eio.Path.open_out ~sw ~create:(`If_missing 0o644) path |> fun f -> (f :> Eio.File.rw_ty Eio.Resource.t) in opening (Writer_lock.file ~writable:true path (file :> Eio.File.ro_ty Eio.Resource.t)) @@ fun () -> claiming ~busy path file @@ fun claim -> recover_journal ~db_page_size:default_page_size ~file (journal_path path); let pager = Btree_eio.Pager.v ~page_size:default_page_size file in let t = handle ~busy ~pager ~file:(Some (file :> Eio.File.ro_ty Eio.Resource.t)) ~writable_file:(Some file) ~read_only:false ~readers:(Some (Readers.of_file file)) ~claim:(Some claim) ~sw:(Some sw) ~db_path:(Some path) ~synchronous ~journal_mode ~clock ~zone ~file_change_seen:0 ~wal_seen:0 ~wal_pages:(Hashtbl.create 64) () in (match journal_mode with `Wal -> lay_out_wal t | `Journal -> ()); sync_dir_of path; claimed t
let in_memory ?clock ?(zone = utc) () = let clock = Option.map stored_clock clock in let pager = Btree.Pager.mem ~page_size:default_page_size () in init_pages pager; let t = handle ~busy:(Writer_lock.busy 0.) ~pager ~file:None ~writable_file:None ~read_only:false ~readers:None ~claim:None ~sw:None ~db_path:None ~synchronous:`Full ~journal_mode:`Wal ~clock ~zone ~file_change_seen:1 ~wal_seen:0 ~wal_pages:(Hashtbl.create 64) () in rebuild_page1 t; t
(* Whether nothing has been written into the database yet: no schema object at all. sqlite3 lets PRAGMA page_size change the width only while that holds -- btree.c's sqlite3BtreeSetPageSize refuses once page 1 has been read -- and afterwards only a VACUUM re-lays the file. *)let is_unwritten t = List.is_empty t.extra_master && List.is_empty t.all_tables
(* Parse sqlite_master into raw table, index, and other entries. The schema is an ordinary table b-tree rooted at page 1, so a multi-page schema walks its interior and leaf children like any other table. *)let read_master t = let raw_tables = ref [] in let raw_indexes = ref [] in let raw_explicit = ref [] in let raw_extra = ref [] in Btree.Table.iter (master_btree t) (fun _rowid payload -> match classify_master_row payload with | Mtable (name, root, sql) -> raw_tables := (name, root, sql) :: !raw_tables | Mautoindex (idx, tbl, root) -> raw_indexes := (idx, tbl, root) :: !raw_indexes | Mindex (name, tbl, root, sql) -> raw_explicit := (name, tbl, root, sql) :: !raw_explicit | Mextra r -> raw_extra := r :: !raw_extra | Mskip -> ()); ( List.rev !raw_tables, List.rev !raw_indexes, List.rev !raw_explicit, List.rev !raw_extra )
(* Where ANALYZE leaves what it measured, and the text sqlite3 declares it with, verbatim: openStatTable (sqlite/src/analyze.c) builds "CREATE TABLE %Q.%s(%s)" over { "sqlite_stat1", "tbl,idx,stat" } -- three untyped columns, no primary key. The text goes into sqlite_master where the other engine reads it, so it is copied rather than composed. *)let stat_table = "sqlite_stat1"let stat_table_sql = "CREATE TABLE sqlite_stat1(tbl,idx,stat)"
(* Put what an ANALYZE recorded back onto the indexes it describes. The rows are data and nothing here can assume this handle wrote them: a bad name, a non-integer stat or a NULL idx each costs that one row only, because a statistic is an estimate and a missing one already has an answer. The row naming no index carries the table's own count, which nothing reads. *)let load_index_stats all_tables = let index_named tbl name = Option.bind (List.find_opt (fun gt -> ident_equal gt.schema.tbl_name tbl) all_tables) (fun gt -> List.find_opt (fun (i : index) -> ident_equal i.name name) gt.indexes) in match List.find_opt (fun gt -> ident_equal gt.schema.tbl_name stat_table) all_tables with | Some { store = Rowid_rows btree; _ } -> Btree.Table.iter btree (fun _rowid payload -> match Btree.Record.decode payload with | [ Text tbl; Text name; Text s ] -> Option.iter (fun (i : index) -> i.stat <- stat_of_string s) (index_named tbl name) | _ -> ()) | Some { store = Pk_index_rows _; _ } | None -> ()
(* (Re)derive the in-memory schema state -- table B-trees, the key cache of every key-value table, extra master rows, rowid cache -- by reading page 1 and the sqlite_master table out of the pager. Called on open, and again on {!refresh} when another handle's commit may have changed the schema or moved a root. *)let load_schema t = let pager = t.pager in (* A database of zero pages is the empty database: its schema table is an empty cursor (btree.c, btreeCursor), and its header reads as the zeros past the end of the file (pager.c, readDbPage). *) let page1, (raw_tables, raw_indexes, raw_explicit, raw_extra) = if Btree.Pager.page_count pager = 0 then (String.make 100 '\000', ([], [], [], [])) else let page1 = Btree.Pager.read pager 1 in Page1.seed_pager pager page1; (page1, read_master t) in let all_tables = List.map (Reconnect.open_table pager ~raw_indexes ~raw_explicit) raw_tables in load_index_stats all_tables; t.all_tables <- all_tables; t.named_tables <- Kv_store.extract_named all_tables; t.extra_master <- raw_extra; t.master_written <- None; (* User version: a signed 32-bit value stored at header byte 60. The Int32 round trip folds the unsigned word to its signed value on 64-bit and is the identity where int is narrower. *) let uv = Btree.Page.u32_be page1 60 in t.user_version <- Int32.to_int (Int32.of_int uv); (* Whatever wrote page 1 last -- this handle, another one, or sqlite3 -- left its counter there, and this handle's next commit has to bump that value rather than one it remembers from before the reload. *) t.change_counter <- Page1.change_counter page1; t.insert_rowids <- Ident.Tbl.create 8
(* The bytes [PRAGMA cache_size] buys at the page size of [t]'s pager. A budget of nothing still buys the pager's floor, as it does in sqlite3, and not the no limit that [0] means to the pager. *)let cache_bytes t = let n = t.cache_size in max 1 (if n >= 0 then n * Btree.Pager.page_size t.pager else -n * 1024)
(* An in-memory pager keeps every page whatever the budget, as sqlite3's [:memory:] database does: there is no file under it to read a page back from. *)let apply_cache_size t = if Option.is_some t.file then Btree.Pager.set_cache_bytes t.pager (cache_bytes t)
(* The pager a handle reads its file through from here, held to the handle's budget: a pager is built again whenever the file moved under the handle. *)let install_pager t pager = t.pager <- pager; apply_cache_size t
(* Replay the committed log frames into the cache and note which pages they covered: those live only in the cache and the log until a checkpoint puts them in the file, so they are pinned against eviction. The log is then cut back to where the replay stopped, so the file ends on a commit frame and the next append can carry the running checksum on from its header. *)let recover_pending pager wp = let pending = Hashtbl.create 64 in let prefix = recover_into_pager pager wp ~note:(fun p -> Hashtbl.replace pending p (); Btree.Pager.pin pager p) in trim_wal wp prefix.bytes; pending
(* The file change counter the {e file} carries, read straight out of its header rather than through the pager: in WAL mode the pager's page 1 is the log's, and what {!refresh} has to notice is the file moving under the log. A file too short to hold the field has no counter to report and answers 0, which is the value a handle that has read nothing starts from. *)let file_change_counter file = let bytes = Page1.change_counter_offset + 4 in if Optint.Int63.to_int (Eio.File.stat file).size < bytes then 0 else begin let buf = Cstruct.create bytes in Eio.File.pread_exact file ~file_offset:Optint.Int63.zero [ buf ]; Page1.change_counter (Cstruct.to_string buf) end
(* The page size the file declares in its first 100 bytes: the pager is sized by it, so it cannot come through the pager. [None] for a file too short, one that is not a database, or one whose width field names no allowed width. *)let db_header file = if Optint.Int63.to_int (Eio.File.stat file).size < 100 then None else begin let buf = Cstruct.create 100 in Eio.File.pread_exact file ~file_offset:Optint.Int63.zero [ buf ]; let header = Cstruct.to_string buf in if String.sub header 0 16 <> magic then None else Some header end
let header_page_size file = Option.bind (db_header file) page_size_of_header_opt
(* Which of the two ways an existing database's commits are made durable, from the file's own header rather than a preference: offsets 18 and 19 record it, and a reader meeting the rollback-journal version never looks at a [-wal] beside the file however many transactions it holds. Read from the file and not through the pager, before recovery. A file that states no version decides nothing. *)let file_journal_mode file = match Option.bind (db_header file) Page1.file_format_of_header with | Some v when v = Page1.legacy_file_format -> Some `Journal | Some v when v = Page1.wal_file_format -> Some `Wal | Some _ | None -> None
(* The width to lay the pager out at before recovery runs. Page 1 states the page size, but page 1 is exactly the page a log or journal may be the only good copy of, so the logs are asked first, each carrying the width independently: the journal in its header, the log in the width of its frames. A journal written before sqlite3 3.5.8 leaves that field zero and so answers nothing; page 1 then decides, which is also the width its records get cut at. *)let recovery_page_size ~file ~path = match journal_page_size (journal_path path) with | Some ps -> ps | None -> ( match header_page_size file with | Some ps -> ps | None -> Option.value ~default:default_page_size (wal_page_size (wal_path path)))
(* A pager over an existing file, with whatever an interrupted commit left already put back: the rollback journal is replayed before the pager is sized, and the committed WAL frames are read into the cache. The WAL stays on disk until physical detach, which checkpoints the returned replayed pages. A pager with no page at all is a file with no commit in it or its log. *)let recovered_pager ~file ~path = (* Before the width is read, let alone a page written: recovery starts by rolling the journal into the database file, and a refusal after that has already changed the file it was refusing to touch. *) check_sidecars ~db:(Some file) path; let page_size = recovery_page_size ~file ~path in recover_journal ~db_page_size:page_size ~file (journal_path path); let pager = Btree_eio.Pager.v ~page_size file in let wal_pages = recover_pending pager (wal_path path) in (pager, wal_pages)
(* A pager over an existing file for a handle that may only read it: the committed state as the file and its logs stand, with nothing put back and nothing cut away.
The pager is [Btree_eio.Pager.Read_only] rather than [Btree_eio.Pager], so a mode-0444 database opens like any other. Both logs are read rather than recovered: {!trim_wal} would cut away commits another handle appended, and replaying the journal would write the file a live writer is rewriting. What the journal has to say is read at each read instead ({!steady_read}).
The file and log are read one after the other. A checkpoint that would fold the one into the other in between is kept out by the caller's WAL_READ_LOCK(0) ({!hold_unreliable}), or, with no index to lock, watched for in this process ({!Durability.Readers}); a writer may still start the log over, which changes its header (walBeginShmUnreliable compares its salts). An open either crossed is taken again. Every page of the log is pinned as it lands: those pages are in the cache and the log and nowhere else until a checkpoint puts them in the file. *)let read_only_pager ~readers ~file ~path = check_sidecars ~db:(Some file) path; let page_size = recovery_page_size ~file ~path in let pager = Btree_eio.Pager.Read_only.v ~page_size file in let wal_pages = Hashtbl.create 64 in let wp = wal_path path in let rec absorb () = let seq = Readers.checkpoints readers in let log = wal_head wp in let stamp = file_change_counter file in Hashtbl.reset wal_pages; Btree.Pager.refresh_from_file pager; let (_ : wal_prefix) = recover_into_pager pager (wal_path path) ~note:(fun p -> Hashtbl.replace wal_pages p (); Btree.Pager.pin pager p) in if Readers.crossed readers seq || not (String.equal log (wal_head wp)) then begin Domain.cpu_relax (); absorb () end else stamp in let stamp = absorb () in (pager, wal_pages, stamp)
(* WAL_READ_LOCK(0) on the WAL-index of the database at [path], for a reader that cannot write the index (walBeginShmUnreliable), waited for as [busy] waits while a checkpoint copies into the file. [None] where this process may read no index there, and only the count of this process stands ({!Durability.Readers}). *)let hold_unreliable ~busy path = match Writer_lock.retry ~busy (fun () -> match Wal_index.begin_unreliable path with | `Held u -> Some (Some u) | `Missing -> Some None | `Busy -> None) with | Some held -> held | None -> Model.fail_busy "database is locked: %a: a checkpoint is copying its log into it" Eio.Path.pp path
(* A pager over the commit a WAL-index session names: the database file with the frames of the log the file does not hold yet laid over it, each pinned, since it is in the cache and the log and nowhere else. The frames are read through the session's log descriptor ({!Wal_index.read_frames}). [make] builds the pager over the file at the log's page size, or the file's own for a log with no frame. *)let index_pager ~file index make = let last = Wal_index.max_frame index in let page_size = if last > 0 then Wal_index.page_size index else Option.value ~default:default_page_size (header_page_size file) in let pager = make ~page_size in let wal_pages = Hashtbl.create 64 in Wal_index.read_frames index ~apply:(fun p data -> Hashtbl.replace wal_pages p (); Btree.Pager.pin pager p; Btree.Pager.put_committed pager p data); if last > 0 then Btree.Pager.set_page_count pager (Wal_index.pages index); (pager, wal_pages)
(* A read-only reader's answer to a hot journal (pager.c, hasHotJournal). *)let refuse_hot_journal path = Model.fail_readonly_rollback "%a has a hot journal: a writer ended part way through a commit, and only \ a writable open rolls it back" Eio.Path.pp (journal_path path)
(* Look at a rollback-journal file before a read: refuse a hot journal until a writable open recovers it -- a live writer's RESERVED claim means its journal is not hot -- and refresh the pager if a commit moved the file since the previous look. SHARED keeps another process from taking EXCLUSIVE. A writer of this process can hold RESERVED beside the reader, but cannot commit until it releases SHARED, so the journal stays live throughout the read. *)let observe_journal t ~reload path file lock = if journal_is_hot ~db:file (journal_path path) && Option.is_some (Writer_lock.idle lock) && not (Writer_lock.reserved_elsewhere lock) then refuse_hot_journal path; let version = file_change_counter file in if reload || version <> t.file_change_seen then begin Btree.Pager.refresh_from_file t.pager; t.file_change_seen <- version; load_schema t end
(* A rollback-journal read holds SHARED, which keeps the database file stable for the read. The first read under a SHARED looks at the file ({!observe_journal}); a later read under the same SHARED, one of a read transaction, finds the file as that one did, so it looks at neither the counter nor the journal again (lockingv3.html, sections 3 and 4.1). *)let journal_read t ~reload path file lock f = if reload || not t.observed then begin observe_journal t ~reload path file lock; t.observed <- Option.is_some t.shared end; f ()
(* A rollback-journal reader checks the file's change counter before it reads; a WAL reader uses the log's committed prefix ({!refresh}). *)let reads_a_moving_file t = t.read_only && t.journal_mode = `Journal && t.db_path <> None
(* Derive the schema under the same SHARED boundary as a later read. *)let load_committed_schema t = match (reads_a_moving_file t, t.db_path, t.file, t.lock_file) with | true, Some path, Some file, Some lock -> journal_read t ~reload:true path file lock (fun () -> ()) | _ -> load_schema t
(* Which mode to open an existing database in. The mode is the file's, not the caller's: page 1 records it, so [asked] is only what the caller says instead. A file recording nothing readable decides nothing, and the mode a new database is created in stands. *)let opening_journal_mode ~asked file = match asked with | Some mode -> mode | None -> Option.value (file_journal_mode file) ~default:`Wal
(* A handle on a database file that is already there: recovered on the way in for a handle that may write, read as it stands for one that may not. Which one follows from the descriptor rather than a flag: the handle may write exactly where the open asked for permission to and got it. *)let opened ~busy ?index ~sw ~synchronous ~asked ~clock ~zone ~file ~writable_file ~claim path = let read_only = Option.is_none writable_file in let journal_mode = opening_journal_mode ~asked file in let readers = Readers.of_file file in (* How far into the log this handle has read, taken before the fold below rather than after it: a length measured after the fold can name frames the fold never saw, and those frames would be behind the mark and never folded. Taken first, the mark can only fall short, which costs a frame folded a second time. *) let wal_seen = wal_size (wal_path path) in (* The version of the file this handle has read is the one its reading held across, which is the read-only open's to report: it is the open that watched for a checkpoint crossing it. A handle that may write recovered the file on the way in and no other handle of ours may be writing it, so what it has read is what the file says once that is done. *) let pager, wal_pages, file_change_seen = match (writable_file, index) with | None, Some index -> check_sidecars ~db:(Some file) path; let pager, wal_pages = index_pager ~file index (fun ~page_size -> Btree_eio.Pager.Read_only.v ~page_size file) in (pager, wal_pages, file_change_counter file) | None, None -> read_only_pager ~readers ~file ~path | Some writable_file, _ -> let pager, wal_pages = recovered_pager ~file:writable_file ~path in (pager, wal_pages, file_change_counter file) in (* A file of no bytes holds the empty database, as sqlite3 reads one (btree.c, lockBtree, which validates page 1 only when the file has pages). It is what a process that ended between creating the file and writing its first page leaves. It is read as zero pages ({!load_schema}) and laid out as {!init} lays out a new file: by the first write in rollback-journal mode, and here in WAL mode. The pager counts whole pages only, where sqlite3 counts a partial page as one (pager.c, pagerPagecount), so a file shorter than a page has a page 1 that is not a database's, and it is refused as sqlite3 refuses it. *) let unwritten = Btree.Pager.page_count pager = 0 in if unwritten && Optint.Int63.to_int (Eio.File.stat file).size > 0 then raise (Not_a_database "file is not a database"); let t = handle ~busy ~pager ~file:(Some file) ~writable_file ~read_only ~readers:(Some readers) ~claim ~sw:(Some sw) ~db_path:(Some path) ~synchronous ~journal_mode ~clock ~zone ~file_change_seen ~wal_seen ~wal_pages () in t.detached_stamp <- Option.bind index (index_stamp ~file); (match (writable_file, journal_mode) with | Some _, `Wal when unwritten -> lay_out_wal t | Some _, (`Wal | `Journal) | None, _ -> ()); load_committed_schema t; t
(* A handle on a database that is there, for reading alone.
It creates no database and writes no database page or recovery log. It reads under SHARED ({!sharing}), as each read does, and in WAL mode at the commit a read mark of the WAL-index holds, which it may create and write as sqlite3's readers do; where the index cannot be written it reads the log's committed frames under WAL_READ_LOCK(0) ({!hold_unreliable}). It holds nothing once open but the index itself ({!hold_index}). *)let open_read_only ~sw ~busy ~synchronous ~asked ~clock ~zone path = let claimed = claim_back_at_switch_end ~sw in let file = open_in_db ~sw path in let lock = Writer_lock.file ~writable:false path file in opening lock @@ fun () -> sharing ~busy path lock @@ fun () -> let open_on index = claimed (opened ?index ~busy ~sw ~synchronous ~asked ~clock ~zone ~file ~writable_file:None ~claim:None path) in match file_journal_mode file with | Some `Wal -> ( match index_session ~sw ~busy path file Wal_index.begin_read with | `Held index -> Fun.protect ~finally:(fun () -> Wal_index.close index) (fun () -> let t = open_on (Some index) in hold_index ~sw t index; t) | `Unwritable -> let held = hold_unreliable ~busy path in Fun.protect ~finally:(fun () -> Option.iter Wal_index.end_unreliable held) (fun () -> open_on None) | `Busy reason -> Model.fail_busy "database is locked: %a: %s" Eio.Path.pp path reason) | Some `Journal | None -> open_on None
(* Whether a writable handle reads and writes the database at [path], whose descriptor is [file], under the WAL-index: a WAL database of the host by its header, with no rollback journal beside it, which only a session holding the file EXCLUSIVE may roll back. A file no other process can open has no index to share. *)let indexed file path = Option.is_some (Eio.Path.native path) && file_journal_mode file = Some `Wal && not (Eio.Path.is_file (journal_path path))
(* A writable handle on a WAL database whose pager its first session builds, from the WAL-index under the WRITE lock ({!attach}). *)let unattached ~busy ~sw ~synchronous ~clock ~zone ~file path = let ro = (file :> Eio.File.ro_ty Eio.Resource.t) in let page_size = Option.value ~default:default_page_size (header_page_size ro) in handle ~busy ~pager:(Btree_eio.Pager.v ~page_size file) ~file:(Some ro) ~writable_file:(Some file) ~read_only:false ~readers:(Some (Readers.of_file file)) ~claim:None ~sw:(Some sw) ~db_path:(Some path) ~synchronous ~journal_mode:`Wal ~clock ~zone ~file_change_seen:(file_change_counter ro) ~wal_seen:0 ~wal_pages:(Hashtbl.create 64) ()
let open_attached ~sw ?(create = true) ?(read_only = false) ?(busy_timeout = 0.) ?(synchronous = `Full) ?journal_mode ?clock ?(zone = utc) path = let clock = Option.map stored_clock clock in let busy = Writer_lock.busy ?clock busy_timeout in (* [create] means "make the database if it is not there", not "replace whatever I cannot read". Only the one answer that says the file is not there creates it; every other filesystem answer is about a database this handle may not open. What rides on the distinction is the log: [init] removes the recovery files before laying a database out, so a create arm taken on a permission error deletes the durable state of a database that was there all along. A database that is there but whose recovery fails is likewise a database that is there, and its schema must not be written over. *) if read_only then open_read_only ~sw ~busy ~synchronous ~asked:journal_mode ~clock ~zone path else let claimed = claim_back_at_switch_end ~sw in match open_out_db ~sw path with | exception Eio.Io (Eio.Fs.E (Eio.Fs.Not_found _), _) when create -> init ~sw ~claimed ~busy ~synchronous ?journal_mode ~clock ~zone path | file when indexed (file :> Eio.File.ro_ty Eio.Resource.t) path -> claimed (unattached ~busy ~sw ~synchronous ~clock ~zone ~file path) | file -> opening (Writer_lock.file ~writable:true path (file :> Eio.File.ro_ty Eio.Resource.t)) @@ fun () -> claiming ~busy path file @@ fun claim -> claimed (opened ~busy ~sw ~synchronous ~asked:journal_mode ~clock ~zone ~file:(file :> Eio.File.ro_ty Eio.Resource.t) ~writable_file:(Some file) ~claim:(Some claim) path)
(* Pick up commits made by other handles before this handle reads stale cached pages or writes on top of them.
What this handle folded in is a pair: how many bytes of the log, and which version of the file under it. The version is the change counter every commit bumps; a size is not a version, because a log can be checkpointed away and grown back to the length this handle last saw. A log that grew under an unmoved file is a plain append, so the new frames fold in. Anything else means the file moved: drop the cache, re-read it, and replay the log. A no-op for in-memory databases, and only safe at a transaction boundary.
A read-only handle watches for a checkpoint folding one reading into the other and takes its readings again where one crossed them. *)let refresh t = let alone () = Option.bind t.claim Writer_lock.alone in match (t.journal_mode, t.db_path, t.file) with (* A rollback-journal handle folds nothing in: its writer rewrites the database file rather than appending to a log, so there is no committed prefix to absorb and nothing that says how far another handle got. A read-only handle in that mode is served by {!steady_read} instead, which drops the cache and reads the file again whenever it has moved. *) | `Journal, _, _ | _, None, _ | _, _, None -> () (* A session on the WAL-index read the commit it is on when it began ({!attach}, {!hold_read}): nothing is folded in under it. *) | `Wal, Some _, Some _ when Option.is_some t.wal -> () (* Nor does one under WAL_READ_LOCK(0), which read the file and its log whole when it began ({!hold_read}), and keeps every checkpoint out. *) | `Wal, Some _, Some _ when Option.is_some t.unreliable -> () (* A claimed handle that has been its file's only handle in this process since it last read the file has missed nothing: the claim keeps every other process's write out (Writer_lock), and its own commits and checkpoints keep what it has seen current. *) | `Wal, Some _, Some _ when Option.is_some t.alone_seen && alone () = t.alone_seen -> () | `Wal, Some path, Some file -> ( let wp = wal_path path in let fold_log () = ignore (recover_into_pager t.pager wp ~note:(fun p -> Hashtbl.replace t.wal_pages p (); Btree.Pager.pin t.pager p) : wal_prefix) in let absorb ~reread = let size = wal_size wp in let stamp = file_change_counter file in if reread || stamp <> t.file_change_seen || size < t.wal_seen then begin (* The file moved under the log, or the log this handle folded in is gone: another handle checkpointed its commits into the file. The cached pages are stale and the pending-checkpoint set names a log that is no longer there, so drop both and rebuild from the file and whatever log now sits beside it. *) Btree.Pager.refresh_from_file t.pager; Hashtbl.clear t.wal_pages; fold_log (); load_schema t end else if size > t.wal_seen then begin fold_log (); load_schema t end; t.wal_seen <- size; t.file_change_seen <- stamp; t.alone_seen <- alone () in match (t.read_only, t.readers) with | true, Some readers -> let rec settle ~reread = let seq = Readers.checkpoints readers in absorb ~reread; if Readers.crossed readers seq then begin Domain.cpu_relax (); settle ~reread:true end in settle ~reread:false | _ -> absorb ~reread:false)
(* An autocommit boundary: no explicit transaction, no open savepoint, and no statement in flight. Each of the three owns a span the handle is in the middle of -- a transaction holds a snapshot another handle's commits must not be folded under, and a statement holds a journal whose pre-images say which pages were clean when it started, which a commit would erase. *)let at_boundary t = t.txn = None && t.savepoints = [] && t.stmt_depth = 0
(* Refresh only at an autocommit boundary; an open explicit transaction holds its snapshot and must not absorb another handle's commits mid-flight. *)let maybe_refresh t = if at_boundary t then refresh t
(* Append each dirty page as a WAL frame and fsync the log. The last frame states the database's size after the commit, which is what makes it the commit: a reader acts on the frames up to and including one that states a size. The file is left untouched until {!checkpoint}, so a commit costs one sequential log fsync. *)let sync_wal t path file = let pages = ref [] in Btree.Pager.iter_dirty t.pager ~f:(fun page_num data -> pages := (page_num, data) :: !pages); (* In page order, as sqlite3 writes a commit's frames (pcache.c, pcacheSortDirtyList), so pages adjacent in the file are adjacent in the log and a checkpoint copies them in one read and one write. *) let pages = List.sort (fun (a, _) (b, _) -> Int.compare a b) !pages in let page_size = Btree.Pager.page_size t.pager in let page_count = Btree.Pager.page_count t.pager in (match t.wal with | Some index -> Wal_index.commit index ~page_size ~page_count pages | None -> (* Without an index, the log is created at the first commit with the database file's mode (sqlite/src/os_unix.c, findCreateFileMode), read from the file this handle has open, since the path may name another file or none by then; a session on an index created it as it began ({!Wal_index.begin_write}). *) let perm = (Eio.File.stat file).perm land 0o777 in t.wal_seen <- append_wal (wal_path path) ~perm ~page_size ~page_count pages); (* The pages are named as the checkpoint's to fold, and pinned, only once the append has returned -- which is once the log holds the commit. An append that raised carried nothing, and a page recorded for it would be a page the next checkpoint writes out of a commit that never happened. *) List.iter (fun (page_num, _) -> Hashtbl.replace t.wal_pages page_num (); Btree.Pager.pin t.pager page_num) pages; Btree.Pager.commit_dirty t.pager
(* Refuse a commit to a database whose path no longer names the file this handle holds, as sqlite3 refuses to open a rollback journal for it with SQLITE_READONLY_DBMOVED (pager.c, databaseIsUnmoved, asked in pager_open_journal): the journal is created by the path and its pre-images read from the descriptor, so it would sit beside another file, or none, and a rollback would put this file's pages into that one. *)let check_unmoved path file = if has_moved ~file path then Model.fail_readonly_dbmoved "attempt to write a readonly database: %a no longer names the file this \ handle opened" Eio.Path.pp path
(* Rollback-journal commit: write the pre-images of the pages this commit overwrites to the journal and fsync it, write the new page content straight into the file and fsync, then delete the journal. The pre-images are read from the file, so the cache need not preserve them. The journal is opened only for a file its path still names ({!check_unmoved}).
Creating the journal is a change to its directory, so it gets the directory barrier before the first page write, the moment the journal starts being the only way back. A power cut in that window would otherwise leave part of an interrupted commit in the file and no journal to undo it.
A failure once the first page may have gone into the file ends the physical session ({!attach}, [physical_failed]): the file is then neither commit, and the hot journal is the only copy of what it overwrote. The cache goes back to the pre-transaction state, so a later commit of this handle would read its pre-images from the half-written file and write them over that journal; the handle refuses every access instead, and its close leaves the journal for the next open to roll back (pager.c, hasHotJournal). *)let sync_journal t path file = let jp = journal_path path in let page_size = Btree.Pager.page_size t.pager in let orig_pages = Optint.Int63.to_int (Eio.File.stat file).size / page_size in let preimages = ref [] in Btree.Pager.iter_dirty t.pager ~f:(fun p _ -> if p <= orig_pages then begin let buf = Cstruct.create page_size in Eio.File.pread_exact file ~file_offset:(Optint.Int63.of_int ((p - 1) * page_size)) [ buf ]; preimages := (p, Cstruct.to_string buf) :: !preimages end); check_unmoved path file; Eio.Path.with_open_out ~create:(`Or_truncate 0o644) jp (fun j -> Eio.File.pwrite_all j ~file_offset:Optint.Int63.zero [ Cstruct.of_string (encode_journal ~page_size ~nonce:journal_nonce ~orig_pages !preimages); ]; Eio.File.sync j; sync_dir_of jp); (* From the first new page in the file to the barrier over the journal's removal this runs to its end, whatever happens to the fiber. A cancellation would stop at the one point the barrier exists for: an unlink a power cut takes back, over a file this handle has gone on committing to. *) Eio.Cancel.protect (fun () -> match Btree.Pager.sync t.pager; Eio.File.sync file; Eio.Path.unlink jp; sync_dir_of jp with | () -> () | exception exn -> t.physical_failed <- true; raise exn)
(* EXCLUSIVE for the commit a rollback-journal write session puts into the file. Its claim is RESERVED to the readers of this process until then, and the commit waits for the last of them up to [busy_timeout], as sqlite3's busy handler waits on unixLock's refusal (os_unix.c, nShared > 1), and is refused as SQLITE_BUSY after that. A commit that wrote no page, or one that goes to a log, takes none. *)let escalate t = match (t.claim, t.journal_mode) with | Some claim, `Journal when Btree.Pager.has_dirty t.pager -> ( match Writer_lock.retry ~busy:t.busy (fun () -> if Writer_lock.escalate claim then Some () else None) with | Some () -> () | None -> Model.fail_busy "database is locked: a reader of this process holds %a" Fmt.(option ~none:(any "the database") Eio.Path.pp) t.db_path) | _ -> ()
(* Make the current commit durable, in whichever journal mode is active. A commit with no dirty pages is a no-op.
Every dirty page goes out, so this may only be called at a commit boundary: inside a transaction it would write that transaction's pages under a commit marker, which is a commit. {!sync} holds the boundary. *)let sync_committed t = if Btree.Pager.has_dirty t.pager then begin (* A read-only handle holding a dirty page never got past a write door, so reaching here at all is a door that forgot to ask. Refusing before the page reaches the file keeps the promise the handle was opened on. *) check_writable t "making a commit durable"; (* A transaction that writes the database bumps the header's change counter, and this is the one place a commit passes through, so the bump happens once per commit rather than once per page-1 rewrite. Every other reader of the file -- another handle of ours, or an sqlite3 holding a cache -- decides from that value alone whether what it has is still current. *) let seen = t.change_counter in t.change_counter <- Page1.next_change_counter seen; rebuild_page1 t; let dirty = ref 0 in Btree.Pager.iter_dirty t.pager ~f:(fun _ _ -> incr dirty); let write () = match (t.sw, t.db_path, t.writable_file, t.journal_mode) with (* A log beside a file of no pages is one the next sqlite3 to open it deletes (pager.c, pagerOpenWalIfPresent), so a WAL database's first page goes into the file, as it does in the rollback-journal write of sqlite3's PRAGMA journal_mode=WAL (btree.c, sqlite3BtreeSetVersion). *) | Some _, Some path, Some file, mode when mode = `Journal || not (has_pages file) -> sync_journal t path file; (* A journal-mode commit puts page 1 straight into the file. *) t.file_change_seen <- t.change_counter | Some _, Some path, Some file, `Wal -> sync_wal t path file | _ -> Btree.Pager.sync t.pager in match write () with | () -> t.pages_written <- t.pages_written + !dirty | exception exn -> (* The commit did not happen, so neither did the version it would state: page 1 goes back to the one this handle is still on, and its pages stay dirty for a later commit. A version moved by a commit that reached no disk would send other readers to reread what they hold. *) t.change_counter <- seen; rebuild_page1 t; raise exn end
(* The most pages one write of a fold carries, so its buffer stays a bounded copy whatever the size of the log. *)let fold_run_pages = 256
(* Write the pages the log names into [file] as the cache holds them, page 1 last. One write per run of adjacent pages, in file order: a log that holds a large row names a long run, and a write per page would be a system call per page. *)let write_log_pages t file = let page_size = Btree.Pager.page_size t.pager in let write first count = let buf = Cstruct.create (count * page_size) in for i = 0 to count - 1 do Cstruct.blit_from_string (Btree.Pager.read t.pager (first + i)) 0 buf (i * page_size) page_size done; Eio.File.pwrite_all file ~file_offset:(Optint.Int63.of_int ((first - 1) * page_size)) [ buf ] in let rec runs = function | [] -> () | first :: rest -> let rec span count = function | p :: rest when p = first + count && count < fold_run_pages -> span (count + 1) rest | rest -> (count, rest) in let count, rest = span 1 rest in write first count; runs rest in runs (List.sort Int.compare (Hashtbl.fold (fun p () acc -> if p <> 1 then p :: acc else acc) t.wal_pages [])); if Hashtbl.mem t.wal_pages 1 then write 1 1
(* Fold the committed WAL into the database file and remove it, so the file holds the full committed state on its own. The content is already in the cache, so pages are written straight from there -- the WAL is not read back.
Page 1 goes in last. It states the file's version, so writing it first would tell a reader the file has moved while the pages that moved are not all in it, and that reader would ask for a page the count names and the file does not hold. Written last, a reader that meets the old version reads pages it already holds, byte for byte the ones this fold is writing. The one fsync below makes the fold durable, before the log that could repeat it is removed. *)(* Remove the WAL-index of the database at [path], if there is one. *)let remove_index path = match Eio.Path.unlink (wal_index_path path) with | () -> () | exception Eio.Io (Eio.Fs.E (Eio.Fs.Not_found _), _) -> ()
let fold_log t path file = (* Marked from beginning to end, so a handle reading the file across the fold can tell that it did ({!Durability.Readers}) and read again. *) let fold () = (* The pages, the removal and the barrier are one step. A cancellation stops a fiber at its next Eio call, and a fold that stopped inside this one would leave the file holding pages of a log it had gone on to unlink, or an unlink a power cut takes back over a file that already holds every page the log named. *) Eio.Cancel.protect @@ fun () -> if Hashtbl.length t.wal_pages > 0 then begin write_log_pages t file; Eio.File.sync file; Hashtbl.clear t.wal_pages; Btree.Pager.unpin_all t.pager end; let wp = wal_path path in (match remove_wal wp with (* Make the removal durable, not just visible to this handle: otherwise the log survives on the disk and the next reader of it replays a log this checkpoint has already folded into the database file. The pages above reached the file first, so the barrier can only ever expose a file that already holds them. *) | Unlinked -> sync_dir_of wp (* Nothing was taken away, so there is nothing about the directory to make durable. A rollback-journal database has no log and every close of one comes through here, and fsyncing a directory that did not change is the most expensive thing such a close would do -- on Darwin the barrier reaches the drive's own cache. *) | No_log | Kept -> ()); (* The WAL-index goes with the log, as the last connection's close removes it (walformat.html, section 2.3.2). *) remove_index path; t.wal_seen <- 0; (* Page 1 went into the file with the counter the last commit left in it, so that is the file's version from here -- and the value another handle will read to learn its own cache is behind. *) t.file_change_seen <- t.change_counter in match t.readers with | Some readers -> Readers.checkpointing readers fold | None -> fold ()
(* A checkpoint, as walCheckpoint runs it in passive mode under the WAL-index: the log's frames up to the least read mark still held go into the file, and the log stays for the next commit to start over. Without an index, the log is folded whole and removed ({!fold_log}). A handle reading the file that the index cannot reach counts the fold ({!Durability.Readers}). *)let checkpoint t = (* Rewriting the database file and removing the log are the two writes a reader must never make: the log it would remove holds the commits another handle appended after this one last read, and the pages it would write are that handle's from before them. *) check_writable t "checkpoint"; match (t.wal, t.db_path, t.writable_file) with | Some index, Some path, Some file when not (Eio.Path.is_file (wal_path path)) -> (* The log was taken from under the handle, as a log unlinked beside a sqlite3 that holds it open still is: every page of it is pinned in the cache, which writes them into the file, and the index names no frame from here. A handle that holds no page of a log, as one opened after the last close folded it does, writes nothing and syncs nothing (wal.c, walCheckpoint syncs only after a backfill). *) Eio.Cancel.protect (fun () -> if Hashtbl.length t.wal_pages > 0 then begin write_log_pages t file; Eio.File.sync file end; Wal_index.forget index); Hashtbl.clear t.wal_pages; Btree.Pager.unpin_all t.pager; t.file_change_seen <- t.change_counter | Some index, _, Some file -> ( let copy () = Wal_index.checkpoint index ~db:file in match match t.readers with | Some readers -> Readers.checkpointing readers copy | None -> copy () with | `Done n when n >= Wal_index.max_frame index -> Hashtbl.clear t.wal_pages; Btree.Pager.unpin_all t.pager; t.file_change_seen <- t.change_counter | `Done _ | `Busy -> ()) | None, Some path, Some file -> fold_log t path file | _ -> ()
(* The log past [wal_autocheckpoint] frames copied into the file and started over, cut to no bytes, waiting as the handle's busy timeout waits for the readers whose marks keep it ({!Wal_index.copy_back}), and what it did kept for the caller to read. Without an index, a passive {!checkpoint}. *)let copy_back t = match (t.wal, t.db_path, t.writable_file) with | Some index, Some path, Some file when Eio.Path.is_file (wal_path path) -> let copy () = Wal_index.copy_back ~wait:t.busy index ~db:file in let r = match t.readers with | Some readers -> Readers.checkpointing readers copy | None -> copy () in t.copy_back <- Some r; if r.Wal_index.copied >= Wal_index.max_frame index then begin Hashtbl.clear t.wal_pages; Btree.Pager.unpin_all t.pager; t.file_change_seen <- t.change_counter end | _ -> checkpoint t
(* The log folded whole into the file and removed with its index, as the last connection's close does (sqlite3WalClose): once every frame is in the file and EXCLUSIVE shows no other process holds the file. The handle's session goes on under that EXCLUSIVE, with no log and no index, until it ends. [false] where a reader's mark or another process kept the log. *)let fold_and_remove t = match (t.wal, t.db_path, t.writable_file, t.lock_file) with | Some index, Some path, Some _, Some lock -> ( checkpoint t; if Wal_index.min_frame index <= Wal_index.max_frame index then false else match Writer_lock.take_alone lock with | None -> false | Some exclusive -> Eio.Cancel.protect (fun () -> let wp = wal_path path in (match remove_wal wp with | Unlinked -> sync_dir_of wp | No_log | Kept -> ()); remove_index path); Wal_index.close index; t.wal <- None; Option.iter Writer_lock.release t.claim; t.claim <- Some exclusive; t.wal_seen <- 0; t.file_change_seen <- t.change_counter; true) | None, Some path, Some file, _ -> fold_log t path file; true | _ -> true
(* [len] bytes of [fd] at [at], or [""] where the file of [size] bytes ends before them. *)let pread_fd fd ~size at len = if at < 0 || at + len > size then "" else begin let buf = Bytes.create len in ignore (Unix.lseek fd at Unix.SEEK_SET : int); let rec fill off = if off < len then match Unix.read fd buf off (len - off) with | 0 -> () | n -> fill (off + n) in fill 0; Bytes.to_string buf end
(* The log's part of {!physical_stamp}: its size, header and last frame header, [Some "absent"] with no log, [None] for one that cannot be read. *)let log_stamp wal = let read fd = let size = (Unix.fstat fd).st_size in let header = pread_fd fd ~size 0 32 in let last = if String.length header < 32 then "" else let page_size = match String.get_int32_be header 8 with | 1l -> 65536 | n -> Int32.to_int n in pread_fd fd ~size (size - (24 + page_size)) 24 in Fmt.str "%d:%s:%s" size header last in match Unix.openfile wal [ Unix.O_RDONLY; Unix.O_CLOEXEC ] 0 with | exception Unix.Unix_error (Unix.ENOENT, _, _) -> Some "absent" | exception Unix.Unix_error _ -> None | fd -> ( match Fun.protect ~finally:(fun () -> Unix.close fd) (fun () -> read fd) with | log -> Some log (* A log that cannot be read is no version: the next session reads it again, and meets whatever is wrong with it. *) | exception Unix.Unix_error _ -> None)
(* What the file and its log looked like, read as SQLite's pager reads the file's version before it trusts its cache (pager.c, dbFileVers): the 16 bytes at offset 24 of the database header and the file's size; and for the log (walformat.html, section 1) its size, its header and its last frame header, whose running checksum chains every frame before it. Any commit or checkpoint another handle made moves one of them. [None] for a file with no name, whose log this cannot read without a switch, and for a log it cannot read. *)let physical_stamp t = match (t.db_path, t.file) with | Some path, Some file -> ( match Eio.Path.native (wal_path path) with | None -> None | Some wal -> let db_size = Optint.Int63.to_int (Eio.File.stat file).size in let db_version = if db_size < 40 then "" else begin let buf = Cstruct.create 16 in Eio.File.pread_exact file ~file_offset:(Optint.Int63.of_int 24) [ buf ]; Cstruct.to_string buf end in Option.map (Fmt.str "%d:%s:%s" db_size db_version) (log_stamp wal)) | _ -> None
(* A logical writable handle attaches an exclusive physical session before touching its file. The cache it kept across the unlocked interval stands only if the file and its log are as its last session left them ({!physical_stamp}); otherwise the pager and schema are rebuilt from the file, with whatever a crash of another process left put back. *)(* The pager and schema of a writable handle rebuilt from the commit its WAL-index session names. *)let rebuild_from_index t file index = let pager, pages = index_pager ~file:(file :> Eio.File.ro_ty Eio.Resource.t) index (fun ~page_size -> Btree_eio.Pager.v ~page_size file) in install_pager t pager; Hashtbl.clear t.wal_pages; Hashtbl.iter (fun p () -> Hashtbl.replace t.wal_pages p ()) pages; t.journal_mode <- `Wal; load_schema t; t.wal_seen <- 0; t.file_change_seen <- file_change_counter file; t.alone_seen <- None; Hashtbl.clear t.plan_cache
(* A write session of a WAL database under its WAL-index, as a sqlite3 connection writes one (walformat.html, section 2.3.2): SHARED on the file, which readers hold too, and the index's WRITE lock, which keeps out every other writer and no reader. The cache stands while the index names the commit the handle's last session left; otherwise the pager is rebuilt from the file and the frames the index names. [false] where the index cannot be written, and the session is the EXCLUSIVE one of {!attach}. *)let attach_indexed t path file lock sw = let claim = match Writer_lock.take_wal_session ~busy:t.busy lock with | Some claim -> claim | None -> Model.fail_busy "database is locked: %a has an active session" Eio.Path.pp path in (* A session that fails once it holds the WRITE lock keeps it and its claim, and refuses every later access, as a failure past {!attach}'s claim does: what the failure left is not known. *) let failed index = t.claim <- Some claim; t.wal <- Some index; t.physical_failed <- true in match index_session ~on_failure:failed ?held:t.index_hold ~sw ~busy:t.busy path file Wal_index.begin_write with | exception e when not t.physical_failed -> Writer_lock.release claim; raise e | `Unwritable -> Writer_lock.release claim; false | `Busy reason -> Writer_lock.release claim; Model.fail_busy "database is locked: %a: %s" Eio.Path.pp path reason | `Held index -> ( t.claim <- Some claim; t.wal <- Some index; hold_index ~sw t index; let stamp = t.detached_stamp in t.detached_stamp <- None; match (* The log is checked where the handle opens it, once for the life of its descriptor, as sqlite3 reads a log's header only when it recovers the index from it (walIndexRecover); a WAL session reads no rollback journal, which {!indexed} keeps out. *) if stamp <> index_stamp ~file index then rebuild_from_index t file index with | () -> true | exception e -> t.physical_failed <- true; raise e)
(* A write session. A WAL handle reads the file's mode and takes the WAL-index's write session ({!attach_indexed}); a rollback-journal handle asks for EXCLUSIVE before it reads anything of the file, so a session that has to wait is waiting from its first step, and the file's mode read under EXCLUSIVE moves the handle to WAL for its next session where another connection changed it. *)let attach ?sw t = if t.physical_failed then failwith "Sqlite: physical session failed; close and reopen the handle"; let sw = match sw with Some _ -> sw | None -> t.sw in match (t.read_only, t.claim, t.db_path, t.writable_file, t.lock_file, sw) with | false, None, Some path, Some file, Some lock, Some sw when t.journal_mode = `Wal && indexed (file :> Eio.File.ro_ty Eio.Resource.t) path && attach_indexed t path file lock sw -> () | false, None, Some path, Some file, Some lock, _ -> ( let claim = match Writer_lock.take_session ~busy:t.busy lock with | Some claim -> claim | None -> Model.fail_busy "database is locked: %a has an active session" Eio.Path.pp path in t.claim <- Some claim; let stamp = t.detached_stamp in t.detached_stamp <- None; match if Option.is_none stamp || stamp <> physical_stamp t then begin let pager, pages = recovered_pager ~file ~path in install_pager t pager; Hashtbl.clear t.wal_pages; Hashtbl.iter (fun p () -> Hashtbl.replace t.wal_pages p ()) pages; (* A file of no pages states no mode, and the handle's stands. *) t.journal_mode <- Option.value (file_journal_mode file) ~default:t.journal_mode; load_schema t; t.wal_seen <- wal_size (wal_path path); t.file_change_seen <- file_change_counter file; t.alone_seen <- None; Hashtbl.clear t.plan_cache end with | () -> () | exception e -> t.physical_failed <- true; raise e) | _ -> ()
(* The end of a write transaction: every commit the handle made is written to the log or the file, and EXCLUSIVE goes back (lockingv3.html, section 7: an idle connection holds no lock). The log stays, as it does between sqlite3's transactions (wal.html, section 3.1); the next session of any handle, in this process or another, reads it back. A failed write retains the claim and refuses further access until the handle is closed. *)let detach t = match t.claim with | Some claim when t.read_depth = 0 && at_boundary t -> ( protecting t @@ fun () -> match sync_committed t; t.alone_seen <- None; t.detached_stamp <- (match (t.wal, t.file) with | Some index, Some file -> index_stamp ~file index | _ -> physical_stamp t); Option.iter Wal_index.close t.wal; t.wal <- None; t.claim <- None; Writer_lock.release claim with | () -> () | exception e -> t.physical_failed <- true; raise e) | _ -> ()
(* Read a WAL database again for a read-only handle at the commit its WAL-index session names, where that is not the one it last read. *)let reread_index t index = match t.file with | Some file when index_stamp ~file index <> t.detached_stamp -> let stamp = index_stamp ~file index in let pager, pages = index_pager ~file index (fun ~page_size -> Btree_eio.Pager.Read_only.v ~page_size file) in install_pager t pager; Hashtbl.clear t.wal_pages; Hashtbl.iter (fun p () -> Hashtbl.replace t.wal_pages p ()) pages; t.journal_mode <- opening_journal_mode ~asked:None file; load_schema t; t.file_change_seen <- file_change_counter file; t.detached_stamp <- stamp; Hashtbl.clear t.plan_cache | Some _ | None -> ()
(* Read a WAL file and its log again for a read-only handle that reaches no WAL-index, where either moved since it last read them ({!physical_stamp}). *)let reread_wal t path = let now = physical_stamp t in if Option.is_none now || now <> t.detached_stamp then match (t.file, t.readers) with | Some file, Some readers -> let pager, pages, stamp = read_only_pager ~file ~readers ~path in install_pager t pager; Hashtbl.clear t.wal_pages; Hashtbl.iter (fun p () -> Hashtbl.replace t.wal_pages p ()) pages; t.journal_mode <- opening_journal_mode ~asked:None file; load_schema t; t.file_change_seen <- stamp; t.wal_seen <- wal_size (wal_path path); t.detached_stamp <- now; Hashtbl.clear t.plan_cache | _ -> ()
(* Say that this handle is reading its database file, so a checkpoint of it defers rather than folding the log underneath the read ({!Durability.Readers}).
A handle that may write does not say it -- the one writer's checkpoint is its own -- and neither does a rollback-journal handle, where there is no log to fold. Kept for as long as the handle holds a state a checkpoint would move under it: the reads, and any transaction or savepoint they leave open.
A read-only handle on a named file takes SHARED first, for the same span: a checkpoint of another process is one this count never reaches, and SHARED keeps that process from writing the file at all ({!take_shared}). *)let hold_read t = attach t; (match (t.read_only, t.shared, t.db_path, t.lock_file) with | true, None, Some path, Some lock -> ( let shared = take_shared ~busy:t.busy path lock in t.shared <- Some shared; match (* A rollback-journal file is read under {!steady_read}, which drops the cache whenever the file's version moved. A WAL file is read at the commit a read mark of its WAL-index holds (walTryBeginRead), or, where the index cannot be written, under its WAL_READ_LOCK(0) ({!hold_unreliable}), again from the file and its log where either moved since this handle last read them ({!physical_stamp}). *) match (t.file, t.sw) with | Some file, _ when file_journal_mode file <> Some `Wal -> () | None, _ -> () | Some _, None -> if at_boundary t then reread_wal t path | Some file, Some sw -> ( match index_session ?held:t.index_hold ~sw ~busy:t.busy path file Wal_index.begin_read with | `Held index -> t.wal <- Some index; hold_index ~sw t index; if at_boundary t then reread_index t index | `Unwritable -> t.unreliable <- hold_unreliable ~busy:t.busy path; if at_boundary t then reread_wal t path | `Busy reason -> Model.fail_busy "database is locked: %a: %s" Eio.Path.pp path reason) with | () -> () | exception e -> Option.iter Wal_index.close t.wal; t.wal <- None; Option.iter Wal_index.end_unreliable t.unreliable; t.unreliable <- None; t.shared <- None; t.observed <- false; Writer_lock.release shared; raise e) | _ -> ()); t.read_depth <- t.read_depth + 1; match t.readers with | Some readers when t.read_only && t.journal_mode = `Wal && not t.live_read -> Readers.enter readers; t.live_read <- true | _ -> ()
(* Give the count back, whatever the handle was in the middle of. Reached from {!close}: a handle that has said it is reading and is then dropped without this holds a checkpoint off for the life of the process. *)let drop_read t = if t.read_only then begin Option.iter Wal_index.close t.wal; t.wal <- None; Option.iter Wal_index.end_unreliable t.unreliable; t.unreliable <- None end; if t.live_read then begin (match t.readers with Some readers -> Readers.leave readers | None -> ()); t.live_read <- false end; match t.shared with | Some lock -> t.shared <- None; t.observed <- false; Writer_lock.release lock | None -> ()
(* Give the count back once there is nothing left holding a state: no read in flight and no transaction or savepoint open. Every door that ends one of those asks. *)let settle_read t = if t.read_depth = 0 && at_boundary t then begin drop_read t; detach t end
(* Leave a read. The count outlives it where the read opened a transaction, and the transaction's own end is then what gives it back. *)let release_read t = t.read_depth <- t.read_depth - 1; settle_read t
(* Read through a public door. The rule the interface states -- a handle absorbs other handles' commits at its next transaction boundary -- is a rule about every door, and a read is a boundary as much as a write: a row answered from a page another handle has replaced is stale, and the handle that then writes from that page commits it back over the replacement. So the read verbs take the write verbs' guard ({!autocommit_op}), and inside an explicit transaction it is a no-op, which holds that transaction's snapshot.
A read-only handle on a rollback-journal database is the one door that does more, because the file itself moves under the read: {!steady_read} takes the read between two observations of the file's version and retakes it where they disagree, at every read and not only at a boundary.
The read is bracketed by the count a checkpoint defers to ({!hold_read}), the refresh included, and the count is given back however the read ends. *)(* The boundary every read takes, and the whole of what a read is on a handle whose file no other handle rewrites under it: absorb what the others committed, run the body, and give back the count a checkpoint of this file defers to -- however the body ended. *)let bounded_read t f = hold_read t; match maybe_refresh t; f () with | v -> release_read t; v | exception exn -> release_read t; raise exn
let read_op t f = (* A read is one of the two things a scan's callback may not do, and every read door is this one, so this is where it is refused ({!Handle.check_not_scanning}). A script comes in through {!read_op_once} and is refused a statement at a time instead. *) check_not_scanning t "read"; entered t "Sqlite read" @@ fun () -> match (reads_a_moving_file t, t.db_path, t.file, t.lock_file) with | true, Some path, Some file, Some lock -> bounded_read t (fun () -> journal_read t ~reload:false path file lock f) | _ -> bounded_read t f
(* Read through a public door whose body cannot be run a second time.
Taking a read again costs nothing where the body left nothing behind, and a script leaves two things: rows already handed to its caller, and the transaction a transaction verb opened. Neither is undone by dropping the page cache, so a second run hands the same rows over twice and meets its own [BEGIN]. What is taken again here is the observation and not the body: the file's version is settled first with an empty body, then the body runs once inside the ordinary boundary. *)let read_op_once t f = entered t "Sqlite script" @@ fun () -> bounded_read t @@ fun () -> (match (reads_a_moving_file t, t.db_path, t.file, t.lock_file) with | true, Some path, Some file, Some lock -> journal_read t ~reload:false path file lock (fun () -> ()) | _ -> ()); f ()
(* Re-lay the database at [page_size]. Reached only on an unwritten one, so re-laying it builds the empty structure again at the new width: the file is truncated, a pager of that width takes over, and page 1 is created afresh. The reserved space belongs to the file, so it carries over.
The log of the discarded layout goes into the file first, through {!checkpoint}: a database states its width in page 1 while a log is read at the width its own header names, and a file and a log that disagree is not a state this engine reads back at all. The page 1 written below is written from this handle's own fields, which the fold has brought up to date.
[Btree.Pager.sync] cannot be used here: it folds the dirty set into the clean cache, which the statement journal open over this pager records pre-images against, so ocaml-btree refuses it while one is open. The pages go out through the file the pager was built on instead. *)let relay_at_page_size t page_size = check_writable t "PRAGMA page_size"; (* Between the truncation and the last page written the file holds no page 1 at all, which is not a database anything opens. A cancellation stops a fiber at its next Eio call, so the re-lay is run where one cannot reach it. *) protecting t @@ fun () -> if not (fold_and_remove t) then Model.fail_busy "database is locked: another connection reads the log the new page size \ would discard"; let reserved = Btree.Pager.reserved t.pager in let pager = match t.writable_file with | Some file -> Eio.File.truncate file Optint.Int63.zero; Btree_eio.Pager.v ~page_size file | None -> Btree.Pager.mem ~page_size () in Btree.Pager.set_reserved pager reserved; init_pages pager; install_pager t pager; (* The statement opened its journal on the pager being discarded, so open the same depth on the replacement: an unmatched close is an error, and a statement that raises after this must still find a journal to roll back. *) for _ = 1 to t.stmt_depth do Btree.Pager.begin_statement pager done; t.all_tables <- []; t.named_tables <- []; t.insert_rowids <- Ident.Tbl.create 8; Hashtbl.reset t.wal_pages; t.wal_seen <- 0; t.file_change_seen <- t.change_counter; rebuild_page1 t; match t.writable_file with | Some file -> let width = Btree.Pager.page_size pager in for p = 1 to Btree.Pager.page_count pager do Eio.File.pwrite_all file ~file_offset:(Optint.Int63.of_int ((p - 1) * width)) [ Cstruct.of_string (Btree.Pager.read pager p) ] done; Eio.File.sync file | None -> ()
(* Whether a handle is inside a read of this database file at this moment, in which case a checkpoint of it would fold the log into the file the read is answering off ({!Durability.Readers}). *)let read_in_flight t = match t.readers with | Some readers when t.journal_mode = `Wal -> Readers.busy readers | Some _ | None -> false
(* The threshold crossing, deferred while another handle is reading this file: the condition holds until a fold clears it, so a later commit asks again. The cost of deferring is the log growing to hold what the reader is reading. A read-only handle passes the threshold but may not act on it, so the test is here rather than in {!checkpoint}, which refuses such a handle by raising.
A fold writes the pages the log names as the cache holds them and removes the log. Under [`Off] the cache also holds commits the log does not, whose pages are only partly among the ones it names: folded as they stand, the file would hold part of a commit with the log that completes it gone, and a process that ends before the next sync leaves a database no open reads. So those commits go into the log first, as sqlite3's are at every commit, and the fold writes the log's last commit whole. Reached only at a commit boundary.
Those commits count as the frames they owe the log, one per dirty page, as sqlite3 counts a commit's frames once it wrote them (sqlite3WalDefaultHook): the log a commit leaves is then the one the threshold reads. *)let maybe_autocheckpoint t = let owed = ref 0 in Btree.Pager.iter_dirty t.pager ~f:(fun _ _ -> incr owed); let frames = !owed + match t.wal with Some index -> Wal_index.max_frame index | None -> 0 in if (not t.read_only) && t.wal_autocheckpoint > 0 && (Hashtbl.length t.wal_pages >= t.wal_autocheckpoint || frames >= t.wal_autocheckpoint) && not (read_in_flight t) then begin sync_committed t; copy_back t end
(* The fsync a commit takes, where the handle asks for one: [`Full]/[`Normal] make the commit durable now (SQLite synchronous=FULL), [`Off] leaves the dirty pages for physical detach to flush. Reached at a commit boundary, which is what lets it write every dirty page. *)let sync_on_commit t = match t.synchronous with `Off -> () | `Normal | `Full -> sync_committed t
(* Durability is a property of committed work, so an open transaction or savepoint gets a no-op: its pages belong to a commit that has not happened, and writing them under a commit marker would be that commit. [stmt_depth] counts for the same reason: a statement holds a journal whose pre-images say which pages were clean when it started, and a commit erases them. *)let sync t = entered t "Sqlite.sync" @@ fun () -> if at_boundary t then begin attach t; sync_committed t; detach t end
(* sqlite3_wal_checkpoint, PASSIVE, as a door of its own: the committed frames copied into the file up to the least read mark, waiting for no reader, and the log started over by the next commit once every frame is in the file. Deferred while a handle of this process reads the file, as {!maybe_autocheckpoint} defers. *)let checkpoint_log t = entered t "Sqlite.checkpoint" @@ fun () -> check_writable t "checkpoint"; if at_boundary t && not (read_in_flight t) then begin attach t; sync_committed t; checkpoint t; detach t end
(* Change how this handle's commits are made durable. The mode is not only the handle's: page 1 records it, and a reader meeting the rollback-journal version never looks at a [-wal] beside the file. So the change is a write.
Leaving [`Wal] folds the log into the file and removes it first, since a log left behind is committed transactions nothing looks at again. Refused inside a transaction or savepoint, in sqlite3's own words (vdbe.c, OP_JournalMode): the mode decides where that transaction's pages are going, and it has already started putting them there. The statement journal is not counted, that being this pragma's own. *)let set_journal_mode t mode = let changed = match (mode, t.journal_mode) with | `Wal, `Journal | `Journal, `Wal -> true | `Wal, `Wal | `Journal, `Journal -> false in if changed then begin (* Only a change is refused. The mode a handle is in is a thing to ask a read-only handle, and asking for the one it is already in changes nothing; moving it rewrites page 1 and folds the log away. *) check_writable t "PRAGMA journal_mode"; if t.txn <> None || t.savepoints <> [] then Fmt.failwith "cannot change %s wal mode from within a transaction" (match mode with `Wal -> "into" | `Journal -> "out of"); if t.journal_mode = `Wal && not (fold_and_remove t) then Model.fail_busy "database is locked: another connection reads the log, so the database \ stays in wal mode"; t.journal_mode <- mode; rebuild_page1 t end
(* Transactions *)
let snapshot_schema t = let table_roots = List.filter_map (fun gt -> match gt.store with | Rowid_rows b -> Some (b, Btree.Table.save_root b) | Pk_index_rows _ -> None) t.all_tables in (* A WITHOUT ROWID table's rows are an index b-tree, whose root moves on a split like any other index's, so its root is saved with the index roots. *) let index_roots = List.concat_map (fun gt -> (match gt.store with | Rowid_rows _ -> [] | Pk_index_rows p -> [ (p.index, Btree.Index.save_root p.index) ]) @ List.map (fun (idx : index) -> (idx.btree, Btree.Index.save_root idx.btree)) gt.indexes) t.all_tables in let rowid_snaps = Ident.Tbl.fold (fun k v acc -> (k, !v) :: acc) t.insert_rowids [] in { rowid_snaps; table_roots; index_roots; all_tables_snap = t.all_tables; named_tables_snap = t.named_tables; extra_master_snap = t.extra_master; user_version_snap = t.user_version; }
let restore_schema t (snap : schema_snapshot) = t.master_written <- None; (* Restore B-tree root pages *) List.iter (fun (btree, root) -> Btree.Table.restore_root btree root) snap.table_roots; List.iter (fun (idx, root) -> Btree.Index.restore_root idx root) snap.index_roots; t.all_tables <- snap.all_tables_snap; (* A key cache goes with its table: DROP TABLE takes the entry out, so an undo of that drop has to hand it back, or the key-value door goes on believing there is no such table until the handle is reopened. *) t.named_tables <- snap.named_tables_snap; t.extra_master <- snap.extra_master_snap; (* The user-version lives in the file header, which the restored pages carry, so the handle's copy of it has to come back with them. *) t.user_version <- snap.user_version_snap; (* Restore insert_rowids *) Ident.Tbl.reset t.insert_rowids; List.iter (fun (name, v) -> Ident.Tbl.replace t.insert_rowids name (ref v)) snap.rowid_snaps; (* The key caches describe pages that have just moved underneath them. *) Kv_store.invalidate_all t
let snapshot_txn t = { pager_snap = Btree.Pager.snapshot t.pager; schema_snap = snapshot_schema t }
let restore_txn t snap = Btree.Pager.rollback t.pager snap.pager_snap; restore_schema t snap.schema_snap
(* The snapshot an unfinished transaction would be undone to: the explicit [BEGIN]'s, else the outermost open savepoint's -- outside a transaction that one defers durability exactly as [BEGIN] does, and the stack grows at the head, so the outermost is its last. *)let open_txn_snapshot t = match t.txn with | Some snap -> Some snap | None -> ( match List.rev t.savepoints with | (_, snap) :: _ -> Some snap | [] -> None)
(* Closing a handle that may only read. It owes the file nothing, so it drops what it held and leaves every byte where it is. Each step a writable close takes would take something from the handle still writing: the checkpoint would put this reader's older pages into the file, removing the log would discard later commits, and two readers closing at once would race to remove the same log. *)let close_read_only t = t.txn <- None; t.savepoints <- []; (* Whatever this handle was in the middle of, it is reading nothing from here: a count it kept would hold a checkpoint of this file off for the life of the process. *) drop_read t; Hashtbl.reset t.wal_pages; Btree.Pager.unpin_all t.pager; t.sw <- None
(* Whether this handle's path no longer names the file it has open (os_unix.c, fileHasMoved). *)let moved t = match (t.db_path, t.file) with | Some path, Some file -> has_moved ~file path | _ -> false
(* Closing a handle that may write.
A transaction or savepoint still open never committed, so it is undone rather than made durable, before the refresh, so the pages discarded are this handle's own. Then pick up other handles' commits, make the commits still in the cache durable as every commit is made, and checkpoint the committed pages and remove the log. A checkpoint that cannot complete is replayed on the next open.
Those last commits go to the log (or the journal), not straight into the file. The log still holds older frames of some of their pages, and until the fold removes it, the next open lays those frames over the file: a process that ended in between would reopen on a mixture of the two, a row pointing at pages that hold another commit's bytes. *)let close_writable t = (* The session's WAL-index is opened on a switch of the close's own: the handle's switch may be ending, which is when it calls this. So is its log, whose descriptor the handle took at its first commit, maybe after the caller arranged this close, and which that switch has closed first then ({!Wal_index.drop_log}). A database in memory makes no Eio call and has no index. *) Option.iter Wal_index.drop_log t.index_hold; let with_switch f = match t.db_path with | Some _ -> Eio.Switch.run (fun sw -> f (Some sw)) | None -> f None in with_switch @@ fun sw -> match if t.claim = None && at_boundary t && not (Btree.Pager.has_dirty t.pager) then attach ?sw t with | exception Model.Busy _ -> (* Another handle is in a transaction on the file: this handle owes it nothing, and the log stays for whichever handle closes last. *) t.sw <- None | () -> attach ?sw t; (match open_txn_snapshot t with | None -> () | Some snap -> restore_txn t snap; t.txn <- None; t.savepoints <- []); refresh t; sync_committed t; (* A handle is reading this file, so the checkpoint would fold the log underneath that read; or the path names another file, or none, so the log and index there are no longer this file's to fold and remove, as sqlite3PagerClose checkpoints only where databaseIsUnmoved. Leave the log: it is the committed state, a later handle folds it in. *) if not (read_in_flight t || moved t) then ignore (fold_and_remove t : bool); detach t; t.sw <- None
(* Clean shutdown that opens no new switch-scoped resource: it writes through the already-open file, the in-memory cache and path operations only, so it is safe to call from [Switch.on_release] while that switch is finishing.
Run where a cancellation cannot reach it, because what it writes is the run: under [`Off] no commit is in the file until this.
Then the descriptor the open took goes back. A handle closed under a switch that lives on would otherwise keep a descriptor until the process reached its limit; Eio's own close removes the switch's release hook, so the later release is a no-op. One close is the whole of it: a writable handle's [file] and [writable_file] are one open read two ways, and closing both would close a descriptor twice, which the in-memory filesystem refuses. A second close is asked to do nothing ({!Handle.is_closed}). *)let close t = if not (is_closed t) then begin protecting t @@ fun () -> if t.physical_failed then begin (* End the failed physical session as a crash would: recovery sidecars remain authority for the next attachment, and no cached page is written out of an uncertain outcome. *) t.txn <- None; t.savepoints <- []; t.sw <- None end else if t.read_only then close_read_only t else close_writable t; (* Closing any descriptor on the file releases every record lock this process holds on it, whichever descriptor took them, so the descriptor goes through {!Writer_lock.close_later}: while a claim of this process stands, it stays open until that claim's last release. This handle's claim is then given back, and whatever claim survives it is taken again for a close that came under no claim. *) Option.iter Writer_lock.close_later t.lock_file; Option.iter Wal_index.close t.wal; t.wal <- None; drop_index_hold t; Option.iter Writer_lock.release t.claim; t.claim <- None; Option.iter Writer_lock.reassert t.lock_file end
let open_ ~sw ?create ?read_only ?busy_timeout ?synchronous ?journal_mode ?clock ?zone path = let t = open_attached ~sw ?create ?read_only ?busy_timeout ?synchronous ?journal_mode ?clock ?zone path in (* A writable handle on a WAL database builds its pager in its first session; one refused gives its descriptor back, as {!opening} does. *) (if (not t.read_only) && Option.is_none t.claim then match attach t with | () -> () | exception e -> Option.iter Writer_lock.close_later t.lock_file; Option.iter Wal_index.close t.wal; t.wal <- None; drop_index_hold t; Option.iter Writer_lock.release t.claim; t.claim <- None; t.sw <- None; raise e); detach t; t
(* The claims a snapshot is captured under: main-file SHARED and a fence over an existing WAL-index, which keeps every writer, checkpoint and recovery off the log while the snapshot takes its read mark ({!Wal_index.pin}). An idle foreign WAL connection keeps SHARED for its lifetime and holds none of the fence's locks. With no index, EXCLUSIVE fences the database and its log instead. *)let snapshot_claims ~sw ~busy path lock = let shared = take_shared ~busy path lock in match Writer_lock.retry ~busy (fun () -> match Wal_index.attach ~sw ~create:false ~perm:0 path with | `Missing -> Some `Missing | `Unwritable -> Some `Unwritable | `Busy -> None | `Attached index -> if Wal_index.fence index then Some (`Held index) else begin Wal_index.close index; None end) with | Some (`Held index) -> ([ shared ], Some index) | Some `Unwritable | None -> Writer_lock.release shared; Model.fail_busy "database is locked: %a has an active WAL session" Eio.Path.pp path | Some `Missing -> ( Writer_lock.release shared; match Writer_lock.take_session ~busy lock with | Some claim -> ([ claim ], None) | None -> Model.fail_busy "database is locked: %a cannot be snapshotted" Eio.Path.pp path) | exception exn -> Writer_lock.release shared; raise exn
(* The committed image of the database [file] at [path], whose lock is [lock]: the file read in one transfer, with the log's committed pages over it. With a read mark of [index] held, those are the frames of the commit the mark names, which a checkpoint does not copy past and a writer does not overwrite; without one, the caller's claims fence the log and it is read whole. The claims stand throughout, so this reads the source and does nothing else; the schema is read from the image once they are given back. A hot journal is refused as a read-only open refuses it ({!observe_journal}). *)let capture_committed_image ?index ~lock ~file path = let pager, wal_pages, file_change_seen = match index with | Some index -> let pager, wal_pages = index_pager ~file index (fun ~page_size -> Btree_eio.Pager.Read_only.v ~page_size file) in (pager, wal_pages, file_change_counter file) | None -> read_only_pager ~readers:(Readers.of_file file) ~file ~path in if journal_is_hot ~db:file (journal_path path) && not (Writer_lock.reserved_elsewhere lock) then refuse_hot_journal path; let page_size = Btree.Pager.page_size pager in let page_count = Btree.Pager.page_count pager in if page_count = 0 && Optint.Int63.to_int (Eio.File.stat file).size > 0 then raise (Not_a_database "file is not a database"); let image = Bytes.make (page_count * page_size) '\x00' in let main_bytes = min (Bytes.length image) (Optint.Int63.to_int (Eio.File.stat file).size) in if main_bytes > 0 then begin let base = Cstruct.create main_bytes in Eio.File.pread_exact file ~file_offset:Optint.Int63.zero [ base ]; Cstruct.blit_to_bytes base 0 image 0 main_bytes end; let overlay page = if page >= 1 && page <= page_count && page <> Btree.Pager.locking_page pager then Bytes.blit_string (Btree.Pager.read pager page) 0 image ((page - 1) * page_size) page_size in (* Pages introduced by the WAL are beyond the main file; committed WAL pages within it supersede the bytes read from the main file, among them every page a checkpoint up to the read mark may be writing meanwhile. *) for page = (main_bytes / page_size) + 1 to page_count do overlay page done; Hashtbl.iter (fun page () -> overlay page) wal_pages; (image, page_size, opening_journal_mode ~asked:None file, file_change_seen)
let open_snapshot ?(busy_timeout = 0.) ?(read_only = true) ?clock ?(zone = utc) path = let clock = Option.map stored_clock clock in let busy = Writer_lock.busy ?clock busy_timeout in let image, page_size, journal_mode, file_change_seen = Eio.Switch.run @@ fun sw -> let file = open_in_db ~sw path in let lock = Writer_lock.file ~writable:false path file in let claims, index = opening lock @@ fun () -> snapshot_claims ~sw ~busy path lock in let pinned = match index with | Some index when Wal_index.pin index -> Some index | Some _ | None -> None in (* The descriptor is given back under the claims, as a close does: closing it releases every record lock this process holds on the file. *) Fun.protect ~finally:(fun () -> Writer_lock.close_later lock; Writer_lock.reassert lock; Option.iter Wal_index.close index; List.iter Writer_lock.release claims) @@ fun () -> capture_committed_image ?index:pinned ~lock ~file path in let pager = Btree.Pager.mem_of_string ~page_size (Bytes.unsafe_to_string image) in let snapshot = handle ~busy ~pager ~file:None ~writable_file:None ~read_only ~readers:None ~claim:None ~sw:None ~db_path:None ~synchronous:`Full ~journal_mode ~clock ~zone ~file_change_seen ~wal_seen:0 ~wal_pages:(Hashtbl.create 1) () in load_committed_schema snapshot; snapshot
(* The pages are written at their offsets and the file cut to the count read, so a read that {!read_op} runs again over a file that moved leaves the copy of the second run alone. The locking page holds no data and is written as zeros, as sqlite3 leaves it. *)let write_pages t out = let pager = t.pager in let page_size = Btree.Pager.page_size pager in let pages = Btree.Pager.page_count pager in let locking = Btree.Pager.locking_page pager in for page = 1 to pages do let bytes = if page = locking then String.make page_size '\x00' else Btree.Pager.read pager page in Eio.File.pwrite_all out ~file_offset:(Optint.Int63.of_int ((page - 1) * page_size)) [ Cstruct.of_string bytes ] done; Eio.File.truncate out (Optint.Int63.of_int (pages * page_size))
let backup t dst = if not (at_boundary t) then invalid_arg "Sqlite.backup: inside a transaction or savepoint"; (* A copy that did not complete is removed: a part of a database is not one. The refusal of an existing [dst] is the open's, so it removes nothing. *) Eio.Switch.run (fun sw -> let out = Eio.Path.open_out ~sw ~create:(`Exclusive 0o644) dst in match read_op t (fun () -> write_pages t out); Eio_fs_sync.sync_file out with | () -> () | exception exn -> let bt = Printexc.get_raw_backtrace () in (try Eio.Path.unlink dst with Eio.Io _ -> ()); Printexc.raise_with_backtrace exn bt); Eio_fs_sync.sync_parent dst
let write_op t f = entered t "Sqlite write" @@ fun () -> bounded_read t f
let with_open ~sw ?create ?read_only ?busy_timeout ?synchronous ?journal_mode ?clock ?zone path f = let db = open_ ~sw ?create ?read_only ?busy_timeout ?synchronous ?journal_mode ?clock ?zone path in Fun.protect ~finally:(fun () -> close db) (fun () -> f db)
(* Run [f] under a statement journal: on the way out of an exception every page it changed goes back, and so does the engine state that describes them. This is SQLite's implicit per-statement savepoint (vdbeaux.c, sqlite3VdbeHalt), and it makes a statement all-or-nothing whether or not a transaction encloses it. The undo is scoped to [f]: an enclosing transaction's snapshot is neither consulted nor disturbed. [keep] is asked, of a statement that raised, whether what it wrote stays: a FAIL resolution stops at the offending row and undoes nothing before it, so its journal is committed. It defaults to undoing. *)let with_stmt_journal ?(keep = fun () -> false) t f = (* The statement journal is the undo of a statement that writes, and nothing else opens one, so this is the door every SQL mutation and every PRAGMA that sets something comes in through. A read-only handle is turned away here, before the statement has touched a page. *) check_writable t "a statement that writes"; Btree.Pager.begin_statement t.pager; t.stmt_depth <- t.stmt_depth + 1; let schema = snapshot_schema t in let close () = t.stmt_depth <- t.stmt_depth - 1 in match f () with | result -> Btree.Pager.commit_statement t.pager; close (); result | exception exn -> if keep () then Btree.Pager.commit_statement t.pager else ( Btree.Pager.rollback_statement t.pager; restore_schema t schema); close (); raise exn
(* Make the transaction that has just ended durable, or leave the handle as if it had never run. [snap] is its rollback snapshot, still in hand: a durability step that raises is a transaction that did not happen, and the pages and schema go back to [snap] before the failure is re-raised. Every transaction boundary comes through here.
What a failure leaves on disk is the journal mode's to say. Under [`Journal] the commit writes and fsyncs the journal before touching a page, so a failure at or before that barrier leaves the file as it was, and one after leaves a half-rewritten file the next open rolls back, with this handle refusing every access until it is closed ({!sync_journal}). Under [`Wal] the frames are cut back off, leaving the log ending on the last durable commit. In memory both modes leave the state held before the transaction ran. *)let commit_durably t snap = (match escalate t; sync_on_commit t with | () -> () | exception exn -> restore_txn t snap; raise exn); maybe_autocheckpoint t
(* Run [f] as the transaction it is, and commit it. A mutating statement outside an explicit transaction is its own transaction, so it is made durable here or it did not happen: what [f] wrote goes back to a rollback point taken before it ran.
Inside an explicit transaction, under a savepoint or inside another statement this is [f] and nothing else: each owns a span whose own end commits it.
The undo is that snapshot and not the statement's journal, because the two cannot be open at once: a commit folds the dirty set into the clean cache, which is what a journal's pre-images are recorded against. The snapshot is taken where that step can raise: under [`Normal] and [`Full], and under a rollback journal, whose commit waits for EXCLUSIVE ({!escalate}). A log under [`Off] has nothing to go back from. *)let autocommit t f = if not (at_boundary t) then f () else let snap = match (t.synchronous, t.journal_mode) with | `Off, `Wal -> None | (`Normal | `Full), _ | `Off, `Journal -> Some (snapshot_txn t) in let r = f () in (match snap with | Some snap -> commit_durably t snap | None -> maybe_autocheckpoint t); r
(* Run [f] as a transaction. Setting [t.txn] suppresses the per-mutation autocommit boundary inside the body, so every mutation commits atomically: the outermost transaction is made durable once, and its rollback snapshot is not let go until that has happened ({!commit_durably}). The previous [t.txn] is saved and restored so nested calls act as savepoints. *)type refusal = Contended of string
let with_transaction_result t f = let exception Own_contention of string in let own operation = match operation () with | value -> value | exception Busy reason -> raise (Own_contention reason) in let callback_failure = ref None in match own (fun () -> hold_read t); Fun.protect ~finally:(fun () -> release_read t) @@ fun () -> let outer = t.txn in if outer = None then own (fun () -> refresh t); let snap = own (fun () -> snapshot_txn t) in t.txn <- Some snap; match f () with | result -> t.txn <- outer; if outer = None then own (fun () -> commit_durably t snap); settle_read t; result | exception exn -> let trace = Printexc.get_raw_backtrace () in callback_failure := Some (exn, trace); restore_txn t snap; t.txn <- outer; settle_read t; raise exn with | result -> Ok result | exception exn when Option.is_some !callback_failure -> ( let trace = Printexc.get_raw_backtrace () in match !callback_failure with | Some (callback, original_trace) -> Printexc.raise_with_backtrace callback original_trace | None -> Printexc.raise_with_backtrace exn trace) | exception Own_contention reason -> Error (Contended reason)
let with_transaction t f = match with_transaction_result t f with | Ok value -> value | Error (Contended reason) -> raise (Busy reason)
(* Explicit transactions and savepoints: the boundaries a rollback restores to. Placed with the rest of the session because each is a mutation of the snapshot state above rather than of the pages. *)
(* Explicit transactions. [BEGIN] saves a rollback snapshot in [t.txn]; [COMMIT] keeps the changes (drops the snapshot); [ROLLBACK] restores it. *)let begin_txn t = match t.txn with | Some _ -> failwith "cannot start a transaction within a transaction" | None -> (* Snapshot a current view: pick up other handles' commits first. *) refresh t; t.txn <- Some (snapshot_txn t)
let commit_txn t = match t.txn with | None -> failwith "cannot commit - no transaction is active" | Some snap -> (* EXCLUSIVE before anything ends: refused, the transaction stays open for a later COMMIT, as vdbe.c OP_AutoCommit leaves it on SQLITE_BUSY. *) escalate t; t.txn <- None; t.savepoints <- []; (* Back at a boundary, so the snapshot this handle was holding is over and a checkpoint of the file may fold under it again. Before the durability step, which is where a checkpoint is reached -- and which is handed the snapshot anyway, this being a COMMIT that has not committed until its writes are down. *) settle_read t; commit_durably t snap
let rollback_txn t = match t.txn with | None -> failwith "cannot rollback - no transaction is active" | Some snap -> restore_txn t snap; t.txn <- None; t.savepoints <- []; settle_read t
(* SAVEPOINT/RELEASE/ROLLBACK TO. Outside an explicit transaction the outermost savepoint behaves like BEGIN (it refreshes to a current view and defers durability until released). *)let exec_savepoint t name = if at_boundary t then refresh t; t.savepoints <- (name, snapshot_txn t) :: t.savepoints
let exec_release t name = let rec drop = function | [] -> Fmt.failwith "no such savepoint: %s" name | (n, snap) :: rest -> if ident_equal n name then (rest, snap) else drop rest in (* The released savepoint's own snapshot is what a release that commits is undone to: everything since it was taken is the transaction being made durable. *) let kept, snap = drop t.savepoints in (* A release that commits takes EXCLUSIVE first, and refused leaves the savepoint open, as vdbe.c OP_Savepoint does on SQLITE_BUSY. *) if t.txn = None && kept = [] && t.stmt_depth = 0 then escalate t; t.savepoints <- kept; (* releasing the outermost savepoint with no enclosing transaction commits *) settle_read t; if at_boundary t then commit_durably t snap
let exec_rollback_to t name = let rec find = function | [] -> Fmt.failwith "no such savepoint: %s" name | (n, snap) :: rest -> if ident_equal n name then ((n, snap) :: rest, snap) else find rest in let kept, snap = find t.savepoints in restore_txn t snap; (* the savepoint itself (and the ones below it) stay open *) t.savepoints <- kept