Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361//! Fills the mailbox, and so the cache, with older mail while the app//! is open: it goes down Gmail's list of all mail from the newest, a//! thread at a time, and reads each thread the mailbox lacks or has an//! older copy of. Where it got to is kept with the mail (Mailbox.walk),//! so the next run goes on from there; once the mailbox holds as much//! as the cache keeps, it reads only threads that changed. The threads//! the walk never passed, nor read since it began, are gone from Gmail's//! list of all mail when it ends, as threads deleted or trashed while//! the history since the last run had expired are: they leave the mail//! at no cost in quota.//!//! Quota: Gmail allows a user 6,000 units a minute. A thread read costs//! 40 and a page of 500 ids 10. The backfill sends one request at a//! time, and takes a step a pace after the last one's answer: a step//! reads one thread, after a page of ids when it needs the next, so 50//! units at most. At 48 steps a minute, that is 2,400 units, two fifths//! of the quota, and some 1,920 as a rule; filling 100,000 threads takes//! about 35 hours of the app being open. It waits while the host is busy//! (the user's requests, which have the rest of the quota), and backs//! off, doubling, when Gmail refuses or fails it.
const std = @import("std");const siv = @import("siv");const Gmail = @import("Gmail.zig");const Mailbox = @import("Mailbox.zig");const Pager = @import("Pager.zig");const cache = @import("cache.zig");const testing = std.testing;const Allocator = std.mem.Allocator;const ArrayList = std.ArrayList;const log = std.log.scoped(.backfill);
const Backfill = @This();
gpa: Allocator,app: siv.App,pager: Pager,host: Host,timing: Timing,/// Until the next step.timer: ?siv.Timer,/// Failures in a row, for the back-off.failures: u8,state: enum { stopped, walking, /// Asked to start from the top while a request was out: it does once /// the answer comes. walking_then_top,},
/// What the backfill fills, and when it may.pub const Host = struct { context: *anyopaque, /// The mailboxes to fill, the first being the one to judge by: the /// desk's, and one waiting to replace it. mailboxes: *const fn (context: *anyopaque, buf: *[2]*Mailbox) []const *Mailbox, /// Whether to wait: the user's own requests come first. busy: *const fn (context: *anyopaque) bool, /// Hears the threads added or changed, as read. filled: *const fn (context: *anyopaque, threads: []const Gmail.Thread) void, /// Hears Gmail's ids of the threads the walk found gone as it ended, /// to take them out of the mailboxes. gone: *const fn (context: *anyopaque, ids: []const []const u8) void,};
pub const Timing = struct { /// Between an answer and the next request. pace_ms: u32, /// The first wait after a failure, doubled for each more in a row. retry_ms: u32,};
/// The app's timing: 48 steps a minute at most, and from a minute to/// half an hour after a failure.pub const default_timing: Timing = .{ .pace_ms = 1250, .retry_ms = 60_000 };
/// The most failures a back-off doubles for: 2^5 minutes, past which the/// walk also starts over, as a page token Gmail has forgotten fails/// every time.const max_doublings = 5;
/// All mail, as the list's first page shows it: what the backfill walks.const all_mail: Gmail.Query = .{ .labels = &.{}, .search = "" };
pub fn init(gpa: Allocator, app: siv.App, gmail: *const Gmail, host: Host, timing: Timing) Backfill { return .{ .gpa = gpa, .app = app, .pager = .init(gpa, gmail), .host = host, .timing = timing, .timer = null, .failures = 0, .state = .stopped, };}
/// Stops. Deinit the Http first, so no reply comes after.pub fn deinit(self: *Backfill) void { self.stop(); self.pager.deinit(); self.* = undefined;}
/// Goes on from where the mailbox says the walk got to, after a pace,/// unless it is done; or from the top when `from` is .top, as after/// the mail was read whole, to see what changed meanwhile.pub fn start(self: *Backfill, from: enum { where_it_was, top }) !void { if (self.pager.busy()) { if (from == .top) self.state = .walking_then_top; return; } self.stop(); var buf: [2]*Mailbox = undefined; const mailboxes = self.host.mailboxes(self.host.context, &buf); if (from == .top) { for (mailboxes) |m| m.startWalk(); } const page = switch (mailboxes[0].walk) { .page => |p| p, .done => return, }; try self.pager.reset(all_mail, page); self.failures = 0; self.state = .walking; self.later(self.timing.pace_ms);}
/// Stops until `start`; an answer on its way is still taken, and asks/// nothing more.pub fn stop(self: *Backfill) void { if (self.timer) |t| siv.timerCancel(t); self.timer = null; self.state = .stopped;}
fn later(self: *Backfill, ms: u32) void { self.timer = siv.timerStart(self.app, ms, onTimer, null, self) catch |err| blk: { log.err("setting the timer: {t}", .{err}); break :blk null; };}
fn onTimer(user_data: ?*anyopaque) callconv(.c) void { const self: *Backfill = @ptrCast(@alignCast(user_data)); self.timer = null; if (self.host.busy(self.host.context)) return self.later(self.timing.pace_ms); var buf: [2]*Mailbox = undefined; const mail = self.host.mailboxes(self.host.context, &buf)[0]; const want: Pager.Want = if (mail.threads.items.len >= cache.kept) .changed else .unknown_or_changed; self.pager.goOn(mail, .{ .reads = 1, .pages = 1, .want = want }, self, onPage) catch |err| self.failed(err);}
fn onPage(context: *anyopaque, result: anyerror!Pager.Page) void { const self: *Backfill = @ptrCast(@alignCast(context)); const page = result catch |err| return self.failed(err); self.take(page) catch |err| return self.failed(err); switch (self.state) { .stopped => return, .walking_then_top => return self.start(.top) catch |err| log.err("starting over: {t}", .{err}), .walking => {}, } self.failures = 0; if (page.more) return self.later(self.timing.pace_ms); self.state = .stopped; log.info("the walk is done", .{});}
/// Puts the threads read in the mailboxes, with how far the walk got/// and what it passed; at the end, tells of those it never did.fn take(self: *Backfill, page: Pager.Page) !void { var buf: [2]*Mailbox = undefined; const mailboxes = self.host.mailboxes(self.host.context, &buf); const walk: Mailbox.Walk = if (page.more) .{ .page = self.pager.resumeFrom() } else .done; for (mailboxes) |m| { try Gmail.put(m, page.threads); for (page.ids) |id| m.passed(id); try m.setWalk(walk); } if (page.threads.len > 0) self.host.filled(self.host.context, page.threads); if (page.more) return; var gone: ArrayList([]const u8) = .empty; defer gone.deinit(self.gpa); try mailboxes[0].unwalked(self.gpa, &gone); if (gone.items.len == 0) return; log.info("{d} threads are gone from Gmail", .{gone.items.len}); self.host.gone(self.host.context, gone.items);}
fn failed(self: *Backfill, err: anyerror) void { switch (self.state) { .stopped => return, .walking_then_top => return self.start(.top) catch |e| log.err("starting over: {t}", .{e}), .walking => {}, } const doublings = @min(self.failures, max_doublings); const wait = self.timing.retry_ms << @intCast(doublings); log.info("{t}; trying again in {d} s", .{ err, wait / 1000 }); if (self.failures == max_doublings) { self.pager.reset(all_mail, "") catch |e| log.err("starting over: {t}", .{e}); } self.failures +|= 1; self.later(wait);}
test "the backfill reads one thread at a time, older than the mailbox's, and keeps its place in the mail" { var fake: Gmail.Fake = undefined; try fake.start(&routes); defer fake.stop(); var mail: Mailbox = try .init(testing.allocator); defer mail.deinit(); // The newest, read already; the walk was left at page 2. _ = try mail.put(.{ .id = "a", .participants = "", .subject = "", .snippet = "", .date = 0, .messages = 1, .labels = &.{}, .unread = false, .notes = &.{}, }, 9); try mail.setWalk(.{ .page = "p2" }); fake.fail_path = "/threads/c?"; fake.fail_count.store(1, .monotonic); var host: TestHost = .{ .mail = &mail, .busy = true, .filled = 0 }; var backfill: Backfill = .init(testing.allocator, fake.app, &fake.gmail, host.host(), .{ .pace_ms = 5, .retry_ms = 10, }); defer backfill.deinit(); try backfill.start(.where_it_was);
// Nothing while the host is busy. for (0..10) |_| try siv.appDispatch(fake.app, 10); try testing.expectEqual(0, fake.requests.load(.monotonic));
// Page 2 and its two threads, then page 3's: c fails once, and is // read on the next try. host.busy = false; var waited: u32 = 0; while (mail.walk != .done and waited < 500) : (waited += 1) try siv.appDispatch(fake.app, 10); try testing.expect(mail.walk == .done); for ([_][]const u8{ "b", "c", "d" }) |id| try testing.expect(mail.findThread(id) != null); try testing.expectEqual(3, host.filled); // One request at a time; a is not read again, and page 3 is asked // again with c. try testing.expectEqual(1, fake.most_out.load(.monotonic)); try testing.expectEqual(7, fake.requests.load(.monotonic));}
test "at the walk's end, the threads it never passed leave the mail, and those read since it began stay" { var fake: Gmail.Fake = undefined; try fake.start(&top_routes); defer fake.stop(); var mail: Mailbox = try .init(testing.allocator); defer mail.deinit(); // From the cache: a is still in Gmail's list, x was trashed or deleted // while the history had expired. The walk went over both before. for ([_][]const u8{ "a", "x" }) |id| _ = try mail.put(testThread(id), 1); try mail.setWalk(.done); var host: TestHost = .{ .mail = &mail, .busy = false, .filled = 0 }; var backfill: Backfill = .init(testing.allocator, fake.app, &fake.gmail, host.host(), .{ .pace_ms = 5, .retry_ms = 10, }); defer backfill.deinit(); try backfill.start(.top);
// Midway, x stays; z comes, as the minute's update reads new mail at // the top of the list, which the walk has passed. var waited: u32 = 0; while (mail.findThread("b") == null and waited < 500) : (waited += 1) try siv.appDispatch(fake.app, 10); try testing.expect(mail.walk != .done and mail.findThread("x") != null); _ = try mail.put(testThread("z"), 9);
while (mail.walk != .done and waited < 500) : (waited += 1) try siv.appDispatch(fake.app, 10); try testing.expect(mail.walk == .done); try testing.expect(mail.findThread("x") == null); for ([_][]const u8{ "a", "b", "c", "z" }) |id| try testing.expect(mail.findThread(id) != null); // Two pages and two threads read: telling x gone costs nothing. try testing.expectEqual(4, fake.requests.load(.monotonic));}
fn testThread(id: []const u8) Mailbox.NewThread { return .{ .id = id, .participants = "", .subject = "", .snippet = "", .date = 0, .messages = 1, .labels = &.{}, .unread = false, .notes = &.{}, };}
/// A host of one mailbox, for the tests.const TestHost = struct { mail: *Mailbox, busy: bool, filled: u32,
fn host(self: *TestHost) Host { return .{ .context = self, .mailboxes = mailboxes, .busy = isBusy, .filled = onFilled, .gone = onGone }; }
fn mailboxes(context: *anyopaque, buf: *[2]*Mailbox) []const *Mailbox { const self: *TestHost = @ptrCast(@alignCast(context)); buf[0] = self.mail; return buf[0..1]; }
fn isBusy(context: *anyopaque) bool { const self: *TestHost = @ptrCast(@alignCast(context)); return self.busy; }
fn onFilled(context: *anyopaque, threads: []const Gmail.Thread) void { const self: *TestHost = @ptrCast(@alignCast(context)); self.filled += @intCast(threads.len); }
fn onGone(context: *anyopaque, ids: []const []const u8) void { const self: *TestHost = @ptrCast(@alignCast(context)); for (ids) |id| self.mail.remove(self.mail.findThread(id).?); }};
const list_path = "/threads?maxResults=500&fields=threads(id,historyId),nextPageToken";
/// Page 2 names a, which the mailbox has as it is, and b; page 3 names c/// and d.const routes = [_]Gmail.Fake.Route{ .ok(list_path ++ "&pageToken=p2", \\{"threads":[{"id":"a","historyId":"9"},{"id":"b","historyId":"8"}],"nextPageToken":"p3"} ), .ok(list_path ++ "&pageToken=p3", \\{"threads":[{"id":"c","historyId":"7"},{"id":"d","historyId":"6"}]} ), .ok("/threads/b?", Gmail.Fake.thread("b", "INBOX", 8)), .ok("/threads/c?", Gmail.Fake.thread("c", "INBOX", 7)), .ok("/threads/d?", Gmail.Fake.thread("d", "INBOX", 6)),};
/// All mail from the top: a, as the mailbox has it, and b on the first/// page; c on the second.const top_routes = [_]Gmail.Fake.Route{ .ok(list_path ++ "&pageToken=p2", \\{"threads":[{"id":"c","historyId":"7"}]} ), .ok(list_path, \\{"threads":[{"id":"a","historyId":"1"},{"id":"b","historyId":"8"}],"nextPageToken":"p2"} ), .ok("/threads/b?", Gmail.Fake.thread("b", "INBOX", 8)), .ok("/threads/c?", Gmail.Fake.thread("c", "INBOX", 7)),};