1- use std:: fs:: { self , File , OpenOptions } ;
2- use std:: io:: { Read , Seek , SeekFrom , Write } ;
3- use std:: path:: { Path , PathBuf } ;
1+ use std:: path:: Path ;
42
53use crate :: wal:: { DecodeError , Frame , ScanResult , ScanStopReason , scan_frames} ;
4+ use allocdb_wal_file:: AppendWalFile ;
65
76#[ derive( Debug ) ]
87pub struct WalFile {
9- path : PathBuf ,
10- file : File ,
8+ file : AppendWalFile ,
119 max_payload_bytes : usize ,
1210}
1311
@@ -43,19 +41,17 @@ impl WalFile {
4341///
4442/// Returns [`WalFileError`] if the file cannot be opened or created.
4543pub fn open ( path : impl AsRef < Path > , max_payload_bytes : usize ) -> Result < Self , WalFileError > {
46- let path = path. as_ref ( ) . to_path_buf ( ) ;
47- let file = open_append_file ( & path) ?;
44+ let file = AppendWalFile :: open ( path) ?;
4845
4946 Ok ( Self {
50- path,
5147 file,
5248 max_payload_bytes,
5349 } )
5450 }
5551
5652 #[ must_use]
5753 pub fn path ( & self ) -> & Path {
58- & self . path
54+ self . file . path ( )
5955 }
6056
6157 /// Appends one encoded frame to the WAL file.
@@ -66,9 +62,7 @@ impl WalFile {
6662/// or [`WalFileError::Io`] if the append fails.
6763pub fn append_frame ( & mut self , frame : & Frame ) -> Result < ( ) , WalFileError > {
6864 self . validate_payload_len ( frame) ?;
69-
70- let encoded = frame. encode ( ) ;
71- self . file . write_all ( & encoded) ?;
65+ self . file . append_bytes ( & frame. encode ( ) ) ?;
7266 Ok ( ( ) )
7367 }
7468
@@ -78,7 +72,7 @@ impl WalFile {
7872///
7973/// Returns [`WalFileError::Io`] if the sync fails.
8074pub fn sync ( & self ) -> Result < ( ) , WalFileError > {
81- self . file . sync_data ( ) ?;
75+ self . file . sync ( ) ?;
8276 Ok ( ( ) )
8377 }
8478
@@ -88,7 +82,7 @@ impl WalFile {
8882///
8983/// Returns [`WalFileError::Io`] if the file cannot be read.
9084pub fn recover ( & self ) -> Result < RecoveredWal , WalFileError > {
91- recover_path ( & self . path )
85+ recover_file ( & self . file )
9286 }
9387
9488 /// Truncates the file to the last valid frame boundary discovered by recovery scanning.
@@ -100,19 +94,16 @@ impl WalFile {
10094/// # Panics
10195///
10296/// Panics only if the discovered valid prefix cannot fit into `u64`.
103- pub fn truncate_to_valid_prefix ( & self ) -> Result < RecoveredWal , WalFileError > {
104- let recovered = recover_path ( & self . path ) ?;
97+ pub fn truncate_to_valid_prefix ( & mut self ) -> Result < RecoveredWal , WalFileError > {
98+ let recovered = recover_file ( & self . file ) ?;
10599 let valid_prefix =
106100 u64:: try_from ( recovered. scan_result . valid_up_to ) . expect ( "valid WAL prefix must fit" ) ;
107101
108102 match recovered. scan_result . stop_reason {
109103 ScanStopReason :: CleanEof => { }
110104 ScanStopReason :: TornTail { .. } => {
111105 if recovered. file_len > valid_prefix {
112- let mut file = OpenOptions :: new ( ) . write ( true ) . open ( & self . path ) ?;
113- file. set_len ( valid_prefix) ?;
114- file. seek ( SeekFrom :: Start ( valid_prefix) ) ?;
115- file. sync_data ( ) ?;
106+ self . file . truncate_to ( valid_prefix) ?;
116107 }
117108 }
118109 ScanStopReason :: Corruption { offset, error } => {
@@ -134,22 +125,11 @@ impl WalFile {
134125 self . validate_payload_len ( frame) ?;
135126 }
136127
137- if let Some ( parent) = self . path . parent ( ) {
138- fs:: create_dir_all ( parent) ?;
139- }
140-
141- let temp_path = self . temp_path ( ) ;
142- {
143- let mut temp_file = File :: create ( & temp_path) ?;
144- for frame in frames {
145- temp_file. write_all ( & frame. encode ( ) ) ?;
146- }
147- temp_file. sync_data ( ) ?;
128+ let mut bytes = Vec :: new ( ) ;
129+ for frame in frames {
130+ bytes. extend_from_slice ( & frame. encode ( ) ) ;
148131 }
149-
150- fs:: rename ( & temp_path, & self . path ) ?;
151- sync_parent_dir ( & self . path ) ?;
152- self . file = open_append_file ( & self . path ) ?;
132+ self . file . replace_with_bytes ( & bytes) ?;
153133 Ok ( ( ) )
154134 }
155135
@@ -163,50 +143,17 @@ impl WalFile {
163143
164144 Ok ( ( ) )
165145 }
166-
167- fn temp_path ( & self ) -> PathBuf {
168- let mut temp_path = self . path . clone ( ) ;
169- let extension = temp_path
170- . extension ( )
171- . and_then ( |value| value. to_str ( ) )
172- . map_or_else ( || "tmp" . to_owned ( ) , |value| format ! ( "{value}.tmp" ) ) ;
173- temp_path. set_extension ( extension) ;
174- temp_path
175- }
176146}
177147
178- fn recover_path ( path : & Path ) -> Result < RecoveredWal , WalFileError > {
179- let mut file = File :: open ( path) ?;
180- let mut bytes = Vec :: new ( ) ;
181- file. read_to_end ( & mut bytes) ?;
148+ fn recover_file ( file : & AppendWalFile ) -> Result < RecoveredWal , WalFileError > {
149+ let bytes = file. read_all ( ) ?;
182150 let scan_result = scan_frames ( & bytes) ;
183151 Ok ( RecoveredWal {
184152 scan_result,
185153 file_len : u64:: try_from ( bytes. len ( ) ) . expect ( "file length must fit into u64" ) ,
186154 } )
187155}
188156
189- fn open_append_file ( path : & Path ) -> Result < File , std:: io:: Error > {
190- OpenOptions :: new ( )
191- . create ( true )
192- . read ( true )
193- . append ( true )
194- . open ( path)
195- }
196-
197- #[ cfg( unix) ]
198- fn sync_parent_dir ( path : & Path ) -> Result < ( ) , std:: io:: Error > {
199- if let Some ( parent) = path. parent ( ) {
200- OpenOptions :: new ( ) . read ( true ) . open ( parent) ?. sync_all ( ) ?;
201- }
202- Ok ( ( ) )
203- }
204-
205- #[ cfg( not( unix) ) ]
206- fn sync_parent_dir ( _path : & Path ) -> Result < ( ) , std:: io:: Error > {
207- Ok ( ( ) )
208- }
209-
210157#[ cfg( test) ]
211158mod tests {
212159 use std:: fs;
0 commit comments