1package main

2

3import (

4 "bytes"

5 "context"

6 "errors"

7 "fmt"

8 "io"

9 "io/fs"

10 "log/slog"

11 "os"

12 "path/filepath"

13 "runtime"

14 "strings"

15 "syscall"

16 "time"

17

18 "github.com/mjl-/bstore"

19

20 "github.com/mjl-/mox/dmarcdb"

21 "github.com/mjl-/mox/mox-"

22 "github.com/mjl-/mox/moxvar"

23 "github.com/mjl-/mox/mtastsdb"

24 "github.com/mjl-/mox/queue"

25 "github.com/mjl-/mox/store"

26 "github.com/mjl-/mox/tlsrptdb"

27)

28

29func backupctl(ctx context.Context, ctl *ctl) {

30 /* protocol:

31 > "backup"

32 > destdir

33 > "verbose" or ""

34 < stream

35 < "ok" or error

36 */

37

38 // Convention in this function: variables containing "src" or "dst" are file system

39 // paths that can be passed to os.Open and such. Variables with dirs/paths without

40 // "src" or "dst" are incomplete paths relative to the source or destination data

41 // directories.

42

43 dstDataDir := ctl.xread()

44 verbose := ctl.xread() == "verbose"

45

46 // Set when an error is encountered. At the end, we warn if set.

47 var incomplete bool

48

49 // We'll be writing output, and logging both to mox and the ctl stream.

50 writer := ctl.writer()

51

52 // Format easily readable output for the user.

53 formatLog := func(prefix, text string, err error, attrs ...slog.Attr) []byte {

54 var b bytes.Buffer

55 fmt.Fprint(&b, prefix)

56 fmt.Fprint(&b, text)

57 if err != nil {

58 fmt.Fprint(&b, ": "+err.Error())

59 }

60 for _, a := range attrs {

61 fmt.Fprintf(&b, "; %s=%v", a.Key, a.Value)

62 }

63 fmt.Fprint(&b, "\n")

64 return b.Bytes()

65 }

66

67 // Log an error to both the mox service as the user running "mox backup".

68 pkglogx := func(prefix, text string, err error, attrs ...slog.Attr) {

69 ctl.log.Errorx(text, err, attrs...)

70

71 _, werr := writer.Write(formatLog(prefix, text, err, attrs...))

72 ctl.xcheck(werr, "write to ctl")

73 }

74

75 // Log an error but don't mark backup as failed.

76 xwarnx := func(text string, err error, attrs ...slog.Attr) {

77 pkglogx("warning: ", text, err, attrs...)

78 }

79

80 // Log an error that causes the backup to be marked as failed. We typically

81 // continue processing though.

82 xerrx := func(text string, err error, attrs ...slog.Attr) {

83 incomplete = true

84 pkglogx("error: ", text, err, attrs...)

85 }

86

87 // If verbose is enabled, log to the cli command. Always log as info level.

88 xvlog := func(text string, attrs ...slog.Attr) {

89 ctl.log.Info(text, attrs...)

90 if verbose {

91 _, werr := writer.Write(formatLog("", text, nil, attrs...))

92 ctl.xcheck(werr, "write to ctl")

93 }

94 }

95

96 if _, err := os.Stat(dstDataDir); err == nil {

97 xwarnx("destination data directory already exists", nil, slog.String("dir", dstDataDir))

98 }

99

100 srcDataDir := filepath.Clean(mox.DataDirPath("."))

101

102 // When creating a file in the destination, we first ensure its directory exists.

103 // We track which directories we created, to prevent needless syscalls.

104 createdDirs := map[string]struct{}{}

105 ensureDestDir := func(dstpath string) {

106 dstdir := filepath.Dir(dstpath)

107 if _, ok := createdDirs[dstdir]; !ok {

108 err := os.MkdirAll(dstdir, 0770)

109 if err != nil {

110 xerrx("creating directory", err)

111 }

112 createdDirs[dstdir] = struct{}{}

113 }

114 }

115

116 // Backup a single file by copying (never hardlinking, the file may change).

117 backupFile := func(path string) {

118 tmFile := time.Now()

119 srcpath := filepath.Join(srcDataDir, path)

120 dstpath := filepath.Join(dstDataDir, path)

121

122 sf, err := os.Open(srcpath)

123 if err != nil {

124 xerrx("open source file (not backed up)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

125 return

126 }

127 defer sf.Close()

128

129 ensureDestDir(dstpath)

130 df, err := os.OpenFile(dstpath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0660)

131 if err != nil {

132 xerrx("creating destination file (not backed up)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

133 return

134 }

135 defer func() {

136 if df != nil {

137 df.Close()

138 }

139 }()

140 if _, err := io.Copy(df, sf); err != nil {

141 xerrx("copying file (not backed up properly)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

142 return

143 }

144 err = df.Close()

145 df = nil

146 if err != nil {

147 xerrx("closing destination file (not backed up properly)", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

148 return

149 }

150 xvlog("backed up file", slog.String("path", path), slog.Duration("duration", time.Since(tmFile)))

151 }

152

153 // Back up the files in a directory (by copying).

154 backupDir := func(dir string) {

155 tmDir := time.Now()

156 srcdir := filepath.Join(srcDataDir, dir)

157 dstdir := filepath.Join(dstDataDir, dir)

158 err := filepath.WalkDir(srcdir, func(srcpath string, d fs.DirEntry, err error) error {

159 if err != nil {

160 xerrx("walking file (not backed up)", err, slog.String("srcpath", srcpath))

161 return nil

162 }

163 if d.IsDir() {

164 return nil

165 }

166 backupFile(srcpath[len(srcDataDir)+1:])

167 return nil

168 })

169 if err != nil {

170 xerrx("copying directory (not backed up properly)", err,

171 slog.String("srcdir", srcdir),

172 slog.String("dstdir", dstdir),

173 slog.Duration("duration", time.Since(tmDir)))

174 return

175 }

176 xvlog("backed up directory", slog.String("dir", dir), slog.Duration("duration", time.Since(tmDir)))

177 }

178

179 // Backup a database by copying it in a readonly transaction.

180 // Always logs on error, so caller doesn't have to, but also returns the error so

181 // callers can see result.

182 backupDB := func(db *bstore.DB, path string) (rerr error) {

183 defer func() {

184 if rerr != nil {

185 xerrx("backing up database", rerr, slog.String("path", path))

186 }

187 }()

188

189 tmDB := time.Now()

190

191 dstpath := filepath.Join(dstDataDir, path)

192 ensureDestDir(dstpath)

193 df, err := os.OpenFile(dstpath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0660)

194 if err != nil {

195 return fmt.Errorf("creating destination file: %v", err)

196 }

197 defer func() {

198 if df != nil {

199 df.Close()

200 }

201 }()

202 err = db.Read(ctx, func(tx *bstore.Tx) error {

203 // Using regular WriteTo seems fine, and fast. It just copies pages.

204 //

205 // bolt.Compact is slower, it writes all key/value pairs, building up new data

206 // structures. My compacted test database was ~60% of original size. Lz4 on the

207 // uncompacted database got it to 14%. Lz4 on the compacted database got it to 13%.

208 // Backups are likely archived somewhere with compression, so we don't compact.

209 //

210 // Tests with WriteTo and os.O_DIRECT were slower than without O_DIRECT, but

211 // probably because everything fit in the page cache. It may be better to use

212 // O_DIRECT when copying many large or inactive databases.

213 _, err := tx.WriteTo(df)

214 return err

215 })

216 if err != nil {

217 return fmt.Errorf("copying database: %v", err)

218 }

219 err = df.Close()

220 df = nil

221 if err != nil {

222 return fmt.Errorf("closing destination database after copy: %v", err)

223 }

224 xvlog("backed up database file", slog.String("path", path), slog.Duration("duration", time.Since(tmDB)))

225 return nil

226 }

227

228 // Try to create a hardlink. Fall back to copying the file (e.g. when on different file system).

229 warnedHardlink := false // We warn once about failing to hardlink.

230 linkOrCopy := func(srcpath, dstpath string) (bool, error) {

231 ensureDestDir(dstpath)

232

233 if err := os.Link(srcpath, dstpath); err == nil {

234 return true, nil

235 } else if os.IsNotExist(err) {

236 // No point in trying with regular copy, we would warn twice.

237 return false, err

238 } else if !warnedHardlink {

239 var hardlinkHint string

240 if runtime.GOOS == "linux" && errors.Is(err, syscall.EXDEV) {

241 hardlinkHint = " (hint: if running under systemd, ReadWritePaths in mox.service may cause multiple mountpoints; consider merging paths into a single parent directory to prevent cross-device/mountpoint hardlinks)"

242 }

243 xwarnx("creating hardlink to message failed, will be doing regular file copies and not warn again"+hardlinkHint, err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

244 warnedHardlink = true

245 }

246

247 // Fall back to copying.

248 sf, err := os.Open(srcpath)

249 if err != nil {

250 return false, fmt.Errorf("open source path %s: %v", srcpath, err)

251 }

252 defer func() {

253 err := sf.Close()

254 ctl.log.Check(err, "closing copied source file")

255 }()

256

257 df, err := os.OpenFile(dstpath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0660)

258 if err != nil {

259 return false, fmt.Errorf("create destination path %s: %v", dstpath, err)

260 }

261 defer func() {

262 if df != nil {

263 err := df.Close()

264 ctl.log.Check(err, "closing partial destination file")

265 }

266 }()

267 if _, err := io.Copy(df, sf); err != nil {

268 return false, fmt.Errorf("coping: %v", err)

269 }

270 err = df.Close()

271 df = nil

272 if err != nil {

273 return false, fmt.Errorf("closing destination file: %v", err)

274 }

275 return false, nil

276 }

277

278 // Start making the backup.

279 tmStart := time.Now()

280

281 ctl.log.Print("making backup", slog.String("destdir", dstDataDir))

282

283 err := os.MkdirAll(dstDataDir, 0770)

284 if err != nil {

285 xerrx("creating destination data directory", err)

286 }

287

288 if err := os.WriteFile(filepath.Join(dstDataDir, "moxversion"), []byte(moxvar.Version), 0660); err != nil {

289 xerrx("writing moxversion", err)

290 }

291 backupDB(dmarcdb.ReportsDB, "dmarcrpt.db")

292 backupDB(dmarcdb.EvalDB, "dmarceval.db")

293 backupDB(mtastsdb.DB, "mtasts.db")

294 backupDB(tlsrptdb.ReportDB, "tlsrpt.db")

295 backupDB(tlsrptdb.ResultDB, "tlsrptresult.db")

296 backupFile("receivedid.key")

297

298 // Acme directory is optional.

299 srcAcmeDir := filepath.Join(srcDataDir, "acme")

300 if _, err := os.Stat(srcAcmeDir); err == nil {

301 backupDir("acme")

302 } else if err != nil && !os.IsNotExist(err) {

303 xerrx("copying acme/", err)

304 }

305

306 // Copy the queue database and all message files.

307 backupQueue := func(path string) {

308 tmQueue := time.Now()

309

310 if err := backupDB(queue.DB, path); err != nil {

311 xerrx("queue not backed up", err, slog.String("path", path), slog.Duration("duration", time.Since(tmQueue)))

312 return

313 }

314

315 dstdbpath := filepath.Join(dstDataDir, path)

316 opts := bstore.Options{MustExist: true, RegisterLogger: ctl.log.Logger}

317 db, err := bstore.Open(ctx, dstdbpath, &opts, queue.DBTypes...)

318 if err != nil {

319 xerrx("open copied queue database", err, slog.String("dstpath", dstdbpath), slog.Duration("duration", time.Since(tmQueue)))

320 return

321 }

322

323 defer func() {

324 if db != nil {

325 err := db.Close()

326 ctl.log.Check(err, "closing new queue db")

327 }

328 }()

329

330 // Link/copy known message files. Warn if files are missing or unexpected

331 // (though a message file could have been removed just now due to delivery, or a

332 // new message may have been queued).

333 tmMsgs := time.Now()

334 seen := map[string]struct{}{}

335 var nlinked, ncopied int

336 err = bstore.QueryDB[queue.Msg](ctx, db).ForEach(func(m queue.Msg) error {

337 mp := store.MessagePath(m.ID)

338 seen[mp] = struct{}{}

339 srcpath := filepath.Join(srcDataDir, "queue", mp)

340 dstpath := filepath.Join(dstDataDir, "queue", mp)

341 if linked, err := linkOrCopy(srcpath, dstpath); err != nil {

342 xerrx("linking/copying queue message", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

343 } else if linked {

344 nlinked++

345 } else {

346 ncopied++

347 }

348 return nil

349 })

350 if err != nil {

351 xerrx("processing queue messages (not backed up properly)", err, slog.Duration("duration", time.Since(tmMsgs)))

352 } else {

353 xvlog("queue message files linked/copied",

354 slog.Int("linked", nlinked),

355 slog.Int("copied", ncopied),

356 slog.Duration("duration", time.Since(tmMsgs)))

357 }

358

359 // Read through all files in queue directory and warn about anything we haven't handled yet.

360 tmWalk := time.Now()

361 srcqdir := filepath.Join(srcDataDir, "queue")

362 err = filepath.WalkDir(srcqdir, func(srcqpath string, d fs.DirEntry, err error) error {

363 if err != nil {

364 xerrx("walking files in queue", err, slog.String("srcpath", srcqpath))

365 return nil

366 }

367 if d.IsDir() {

368 return nil

369 }

370 p := srcqpath[len(srcqdir)+1:]

371 if _, ok := seen[p]; ok {

372 return nil

373 }

374 if p == "index.db" {

375 return nil

376 }

377 qp := filepath.Join("queue", p)

378 xwarnx("backing up unrecognized file in queue directory", nil, slog.String("path", qp))

379 backupFile(qp)

380 return nil

381 })

382 if err != nil {

383 xerrx("walking queue directory (not backed up properly)", err, slog.String("dir", "queue"), slog.Duration("duration", time.Since(tmWalk)))

384 } else {

385 xvlog("walked queue directory", slog.Duration("duration", time.Since(tmWalk)))

386 }

387

388 xvlog("queue backed finished", slog.Duration("duration", time.Since(tmQueue)))

389 }

390 backupQueue(filepath.FromSlash("queue/index.db"))

391

392 backupAccount := func(acc *store.Account) {

393 defer acc.Close()

394

395 tmAccount := time.Now()

396

397 // Copy database file.

398 dbpath := filepath.Join("accounts", acc.Name, "index.db")

399 err := backupDB(acc.DB, dbpath)

400 if err != nil {

401 xerrx("copying account database", err, slog.String("path", dbpath), slog.Duration("duration", time.Since(tmAccount)))

402 }

403

404 // todo: should document/check not taking a rlock on account.

405

406 // Copy junkfilter files, if configured.

407 if jf, _, err := acc.OpenJunkFilter(ctx, ctl.log); err != nil {

408 if !errors.Is(err, store.ErrNoJunkFilter) {

409 xerrx("opening junk filter for account (not backed up)", err)

410 }

411 } else {

412 db := jf.DB()

413 jfpath := filepath.Join("accounts", acc.Name, "junkfilter.db")

414 backupDB(db, jfpath)

415 bloompath := filepath.Join("accounts", acc.Name, "junkfilter.bloom")

416 backupFile(bloompath)

417 db = nil

418 err := jf.Close()

419 ctl.log.Check(err, "closing junkfilter")

420 }

421

422 dstdbpath := filepath.Join(dstDataDir, dbpath)

423 opts := bstore.Options{MustExist: true, RegisterLogger: ctl.log.Logger}

424 db, err := bstore.Open(ctx, dstdbpath, &opts, store.DBTypes...)

425 if err != nil {

426 xerrx("open copied account database", err, slog.String("dstpath", dstdbpath), slog.Duration("duration", time.Since(tmAccount)))

427 return

428 }

429

430 defer func() {

431 if db != nil {

432 err := db.Close()

433 ctl.log.Check(err, "close account database")

434 }

435 }()

436

437 // Link/copy known message files. Warn if files are missing or unexpected (though a

438 // message file could have been added just now due to delivery, or a message have

439 // been removed).

440 tmMsgs := time.Now()

441 seen := map[string]struct{}{}

442 var nlinked, ncopied int

443 err = bstore.QueryDB[store.Message](ctx, db).FilterEqual("Expunged", false).ForEach(func(m store.Message) error {

444 mp := store.MessagePath(m.ID)

445 seen[mp] = struct{}{}

446 amp := filepath.Join("accounts", acc.Name, "msg", mp)

447 srcpath := filepath.Join(srcDataDir, amp)

448 dstpath := filepath.Join(dstDataDir, amp)

449 if linked, err := linkOrCopy(srcpath, dstpath); err != nil {

450 xerrx("linking/copying account message", err, slog.String("srcpath", srcpath), slog.String("dstpath", dstpath))

451 } else if linked {

452 nlinked++

453 } else {

454 ncopied++

455 }

456 return nil

457 })

458 if err != nil {

459 xerrx("processing account messages (not backed up properly)", err, slog.Duration("duration", time.Since(tmMsgs)))

460 } else {

461 xvlog("account message files linked/copied",

462 slog.Int("linked", nlinked),

463 slog.Int("copied", ncopied),

464 slog.Duration("duration", time.Since(tmMsgs)))

465 }

466

467 // Read through all files in account directory and warn about anything we haven't handled yet.

468 tmWalk := time.Now()

469 srcadir := filepath.Join(srcDataDir, "accounts", acc.Name)

470 err = filepath.WalkDir(srcadir, func(srcapath string, d fs.DirEntry, err error) error {

471 if err != nil {

472 xerrx("walking files in account", err, slog.String("srcpath", srcapath))

473 return nil

474 }

475 if d.IsDir() {

476 return nil

477 }

478 p := srcapath[len(srcadir)+1:]

479 l := strings.Split(p, string(filepath.Separator))

480 if l[0] == "msg" {

481 mp := filepath.Join(l[1:]...)

482 if _, ok := seen[mp]; ok {

483 return nil

484 }

485 }

486 switch p {

487 case "index.db", "junkfilter.db", "junkfilter.bloom":

488 return nil

489 }

490 ap := filepath.Join("accounts", acc.Name, p)

491 if strings.HasPrefix(p, "msg"+string(filepath.Separator)) {

492 xwarnx("backing up unrecognized file in account message directory (should be moved away)", nil, slog.String("path", ap))

493 } else {

494 xwarnx("backing up unrecognized file in account directory", nil, slog.String("path", ap))

495 }

496 backupFile(ap)

497 return nil

498 })

499 if err != nil {

500 xerrx("walking account directory (not backed up properly)", err, slog.String("srcdir", srcadir), slog.Duration("duration", time.Since(tmWalk)))

501 } else {

502 xvlog("walked account directory", slog.Duration("duration", time.Since(tmWalk)))

503 }

504

505 xvlog("account backup finished", slog.String("dir", filepath.Join("accounts", acc.Name)), slog.Duration("duration", time.Since(tmAccount)))

506 }

507

508 // For each configured account, open it, make a copy of the database and

509 // hardlink/copy the messages. We track the accounts we handled, and skip the

510 // account directories when handling "all other files" below.

511 accounts := map[string]struct{}{}

512 for _, accName := range mox.Conf.Accounts() {

513 acc, err := store.OpenAccount(ctl.log, accName)

514 if err != nil {

515 xerrx("opening account for copying (will try to copy as regular files later)", err, slog.String("account", accName))

516 continue

517 }

518 accounts[accName] = struct{}{}

519 backupAccount(acc)

520 }

521

522 // Copy all other files, that aren't part of the known files, databases, queue or accounts.

523 tmWalk := time.Now()

524 err = filepath.WalkDir(srcDataDir, func(srcpath string, d fs.DirEntry, err error) error {

525 if err != nil {

526 xerrx("walking path", err, slog.String("path", srcpath))

527 return nil

528 }

529

530 if srcpath == srcDataDir {

531 return nil

532 }

533 p := srcpath[len(srcDataDir)+1:]

534 if p == "queue" || p == "acme" || p == "tmp" {

535 return fs.SkipDir

536 }

537 l := strings.Split(p, string(filepath.Separator))

538 if len(l) >= 2 && l[0] == "accounts" {

539 name := l[1]

540 if _, ok := accounts[name]; ok {

541 return fs.SkipDir

542 }

543 }

544

545 // Only files are explicitly backed up.

546 if d.IsDir() {

547 return nil

548 }

549

550 switch p {

551 case "dmarcrpt.db", "dmarceval.db", "mtasts.db", "tlsrpt.db", "tlsrptresult.db", "receivedid.key", "ctl":

552 // Already handled.

553 return nil

554 case "lastknownversion": // Optional file, not yet handled.

555 default:

556 xwarnx("backing up unrecognized file", nil, slog.String("path", p))

557 }

558 backupFile(p)

559 return nil

560 })

561 if err != nil {

562 xerrx("walking other files (not backed up properly)", err, slog.Duration("duration", time.Since(tmWalk)))

563 } else {

564 xvlog("walking other files finished", slog.Duration("duration", time.Since(tmWalk)))

565 }

566

567 xvlog("backup finished", slog.Duration("duration", time.Since(tmStart)))

568

569 writer.xclose()

570

571 if incomplete {

572 ctl.xwrite("errors were encountered during backup")

573 } else {

574 ctl.xwriteok()

575 }

576}

577