1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
// Copyright 2016 Alex Regueiro
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//

use crate::env::Env;
use crate::{db::DBInner, ffi, ffi_util::to_cpath, DBCommon, Error, ThreadMode};

use libc::c_uchar;
use std::ffi::CString;
use std::path::Path;

/// Represents information of a backup including timestamp of the backup
/// and the size (please note that sum of all backups' sizes is bigger than the actual
/// size of the backup directory because some data is shared by multiple backups).
/// Backups are identified by their always-increasing IDs.
pub struct BackupEngineInfo {
    /// Timestamp of the backup
    pub timestamp: i64,
    /// ID of the backup
    pub backup_id: u32,
    /// Size of the backup
    pub size: u64,
    /// Number of files related to the backup
    pub num_files: u32,
}

pub struct BackupEngine {
    inner: *mut ffi::rocksdb_backup_engine_t,
    _outlive: Env,
}

pub struct BackupEngineOptions {
    inner: *mut ffi::rocksdb_backup_engine_options_t,
}

pub struct RestoreOptions {
    inner: *mut ffi::rocksdb_restore_options_t,
}

impl BackupEngine {
    /// Open a backup engine with the specified options and RocksDB Env.
    pub fn open(opts: &BackupEngineOptions, env: &Env) -> Result<Self, Error> {
        let be: *mut ffi::rocksdb_backup_engine_t;
        unsafe {
            be = ffi_try!(ffi::rocksdb_backup_engine_open_opts(
                opts.inner,
                env.0.inner
            ));
        }

        if be.is_null() {
            return Err(Error::new("Could not initialize backup engine.".to_owned()));
        }

        Ok(Self {
            inner: be,
            _outlive: env.clone(),
        })
    }

    /// Captures the state of the database in the latest backup.
    ///
    /// Note: no flush before backup is performed. User might want to
    /// use `create_new_backup_flush` instead.
    pub fn create_new_backup<T: ThreadMode, D: DBInner>(
        &mut self,
        db: &DBCommon<T, D>,
    ) -> Result<(), Error> {
        self.create_new_backup_flush(db, false)
    }

    /// Captures the state of the database in the latest backup.
    ///
    /// Set flush_before_backup=true to avoid losing unflushed key/value
    /// pairs from the memtable.
    pub fn create_new_backup_flush<T: ThreadMode, D: DBInner>(
        &mut self,
        db: &DBCommon<T, D>,
        flush_before_backup: bool,
    ) -> Result<(), Error> {
        unsafe {
            ffi_try!(ffi::rocksdb_backup_engine_create_new_backup_flush(
                self.inner,
                db.inner.inner(),
                c_uchar::from(flush_before_backup),
            ));
            Ok(())
        }
    }

    pub fn purge_old_backups(&mut self, num_backups_to_keep: usize) -> Result<(), Error> {
        unsafe {
            ffi_try!(ffi::rocksdb_backup_engine_purge_old_backups(
                self.inner,
                num_backups_to_keep as u32,
            ));
            Ok(())
        }
    }

    /// Restore from the latest backup
    ///
    /// # Arguments
    ///
    /// * `db_dir` - A path to the database directory
    /// * `wal_dir` - A path to the wal directory
    /// * `opts` - Restore options
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use rocksdb::backup::{BackupEngine, BackupEngineOptions};
    /// let backup_opts = BackupEngineOptions::default();
    /// let mut backup_engine = BackupEngine::open(&backup_opts, &backup_path).unwrap();
    /// let mut restore_option = rocksdb::backup::RestoreOptions::default();
    /// restore_option.set_keep_log_files(true); /// true to keep log files
    /// if let Err(e) = backup_engine.restore_from_latest_backup(&db_path, &wal_dir, &restore_option) {
    ///     error!("Failed to restore from the backup. Error:{:?}", e);
    ///     return Err(e.to_string());
    ///  }
    /// ```

    pub fn restore_from_latest_backup<D: AsRef<Path>, W: AsRef<Path>>(
        &mut self,
        db_dir: D,
        wal_dir: W,
        opts: &RestoreOptions,
    ) -> Result<(), Error> {
        let c_db_dir = to_cpath(db_dir)?;
        let c_wal_dir = to_cpath(wal_dir)?;

        unsafe {
            ffi_try!(ffi::rocksdb_backup_engine_restore_db_from_latest_backup(
                self.inner,
                c_db_dir.as_ptr(),
                c_wal_dir.as_ptr(),
                opts.inner,
            ));
        }
        Ok(())
    }

    /// Restore from a specified backup
    ///
    /// The specified backup id should be passed in as an additional parameter.
    pub fn restore_from_backup<D: AsRef<Path>, W: AsRef<Path>>(
        &mut self,
        db_dir: D,
        wal_dir: W,
        opts: &RestoreOptions,
        backup_id: u32,
    ) -> Result<(), Error> {
        let c_db_dir = to_cpath(db_dir)?;
        let c_wal_dir = to_cpath(wal_dir)?;

        unsafe {
            ffi_try!(ffi::rocksdb_backup_engine_restore_db_from_backup(
                self.inner,
                c_db_dir.as_ptr(),
                c_wal_dir.as_ptr(),
                opts.inner,
                backup_id,
            ));
        }
        Ok(())
    }

    /// Checks that each file exists and that the size of the file matches our
    /// expectations. it does not check file checksum.
    ///
    /// If this BackupEngine created the backup, it compares the files' current
    /// sizes against the number of bytes written to them during creation.
    /// Otherwise, it compares the files' current sizes against their sizes when
    /// the BackupEngine was opened.
    pub fn verify_backup(&self, backup_id: u32) -> Result<(), Error> {
        unsafe {
            ffi_try!(ffi::rocksdb_backup_engine_verify_backup(
                self.inner, backup_id,
            ));
        }
        Ok(())
    }

    /// Get a list of all backups together with information on timestamp of the backup
    /// and the size (please note that sum of all backups' sizes is bigger than the actual
    /// size of the backup directory because some data is shared by multiple backups).
    /// Backups are identified by their always-increasing IDs.
    ///
    /// You can perform this function safely, even with other BackupEngine performing
    /// backups on the same directory
    pub fn get_backup_info(&self) -> Vec<BackupEngineInfo> {
        unsafe {
            let i = ffi::rocksdb_backup_engine_get_backup_info(self.inner);

            let n = ffi::rocksdb_backup_engine_info_count(i);

            let mut info = Vec::with_capacity(n as usize);
            for index in 0..n {
                info.push(BackupEngineInfo {
                    timestamp: ffi::rocksdb_backup_engine_info_timestamp(i, index),
                    backup_id: ffi::rocksdb_backup_engine_info_backup_id(i, index),
                    size: ffi::rocksdb_backup_engine_info_size(i, index),
                    num_files: ffi::rocksdb_backup_engine_info_number_files(i, index),
                });
            }

            // destroy backup info object
            ffi::rocksdb_backup_engine_info_destroy(i);

            info
        }
    }
}

impl BackupEngineOptions {
    /// Initializes `BackupEngineOptions` with the directory to be used for storing/accessing the
    /// backup files.
    pub fn new<P: AsRef<Path>>(backup_dir: P) -> Result<Self, Error> {
        let backup_dir = backup_dir.as_ref();
        let c_backup_dir = CString::new(backup_dir.to_string_lossy().as_bytes()).map_err(|_| {
            Error::new(
                "Failed to convert backup_dir to CString \
                     when constructing BackupEngineOptions"
                    .to_owned(),
            )
        })?;

        unsafe {
            let opts = ffi::rocksdb_backup_engine_options_create(c_backup_dir.as_ptr());
            assert!(!opts.is_null(), "Could not create RocksDB backup options");

            Ok(Self { inner: opts })
        }
    }

    /// Sets the number of operations (such as file copies or file checksums) that `RocksDB` may
    /// perform in parallel when executing a backup or restore.
    ///
    /// Default: 1
    pub fn set_max_background_operations(&mut self, max_background_operations: i32) {
        unsafe {
            ffi::rocksdb_backup_engine_options_set_max_background_operations(
                self.inner,
                max_background_operations,
            );
        }
    }
}

impl RestoreOptions {
    /// Sets `keep_log_files`. If true, restore won't overwrite the existing log files in wal_dir.
    /// It will also move all log files from archive directory to wal_dir. Use this option in
    /// combination with BackupEngineOptions::backup_log_files = false for persisting in-memory
    /// databases.
    ///
    /// Default: false
    pub fn set_keep_log_files(&mut self, keep_log_files: bool) {
        unsafe {
            ffi::rocksdb_restore_options_set_keep_log_files(self.inner, i32::from(keep_log_files));
        }
    }
}

impl Default for RestoreOptions {
    fn default() -> Self {
        unsafe {
            let opts = ffi::rocksdb_restore_options_create();
            assert!(!opts.is_null(), "Could not create RocksDB restore options");

            Self { inner: opts }
        }
    }
}

impl Drop for BackupEngine {
    fn drop(&mut self) {
        unsafe {
            ffi::rocksdb_backup_engine_close(self.inner);
        }
    }
}

impl Drop for BackupEngineOptions {
    fn drop(&mut self) {
        unsafe {
            ffi::rocksdb_backup_engine_options_destroy(self.inner);
        }
    }
}

impl Drop for RestoreOptions {
    fn drop(&mut self) {
        unsafe {
            ffi::rocksdb_restore_options_destroy(self.inner);
        }
    }
}