Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441//! Keeps every message of a Gmail account for search (Corpus.zig): it//! goes down Gmail's list of every message but drafts, newest first, a//! page of 500 ids at a time, and reads whole each message the corpus//! lacks, or its thread whole when the corpus lacks more than one of it.//! It starts from the top each run, as the ids the corpus has cost only//! their page; once a walk has gone all the way down, a walk ends at the//! first page the corpus has all of.//!//! Quota: a page costs 5 units, a message read whole 60, and a thread//! 120. The walk shares the backfill's two fifths of Gmail's 6,000 a//! minute: it waits while the backfill walks (the host says so), sends//! one request at a time, and waits 25 ms for each unit of the last//! before the next, so 2,400 units a minute at most; 100,000 messages//! take about a day and a half of the app being open. It waits while the//! host is busy with 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 Corpus = @import("Corpus.zig");const Gmail = @import("Gmail.zig");const Http = @import("Http.zig");const testing = std.testing;const Allocator = std.mem.Allocator;const ArenaAllocator = std.heap.ArenaAllocator;const log = std.log.scoped(.harvest);
const Harvest = @This();
gpa: Allocator,app: siv.App,gmail: *const Gmail,corpus: *Corpus,host: Host,timing: Timing,/// Until the next step.timer: ?siv.Timer,/// Failures in a row, for the back-off.failures: u8,walking: bool,/// Whether to walk again from the top once this walk ends.again: bool,/// The next page's token, empty for the first; null after the last page.next: ?[]const u8,/// The page's messages, and the threads of those not kept yet, in the/// order they were listed; `page_arena` holds them and `next`.page: []const []const u8,wanted: []const Want,page_arena: ArenaAllocator,/// Holds the request out.request_arena: ArenaAllocator,/// How many messages this walk kept.kept: u32,
/// When the walk may go on.pub const Host = struct { context: *anyopaque, /// Whether to wait: for the user's requests, for the backfill, and /// for a new access token. busy: *const fn (context: *anyopaque) bool, /// Gmail refused the access token, which has expired: get another. refused: *const fn (context: *anyopaque) void,};
pub const Timing = struct { /// After an answer, for each quota unit its request cost. unit_ms: u32, /// Between looks while the host is busy. busy_ms: u32, /// The first wait after a failure, doubled for each more in a row. retry_ms: u32,};
/// The app's timing: 40 units a second, two fifths of the quota.pub const default_timing: Timing = .{ .unit_ms = 25, .busy_ms = 1250, .retry_ms = 60_000 };
/// A thread whose messages the corpus lacks, and the first of them.const Want = struct { thread: []const u8, first: []const u8, messages: u32 };
/// What each request costs in quota units.const page_units = 5;const message_units = 60;const thread_units = 120;
/// 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;
/// How many messages kept between the walk's notes in the log.const note_every = 1000;
pub fn init(gpa: Allocator, app: siv.App, gmail: *const Gmail, corpus: *Corpus, host: Host, timing: Timing) Harvest { return .{ .gpa = gpa, .app = app, .gmail = gmail, .corpus = corpus, .host = host, .timing = timing, .timer = null, .failures = 0, .walking = false, .again = false, .next = null, .page = &.{}, .wanted = &.{}, .page_arena = .init(gpa), .request_arena = .init(gpa), .kept = 0, };}
/// Stops. Deinit the Http first, so no reply comes after.pub fn deinit(self: *Harvest) void { self.stop(); self.page_arena.deinit(); self.request_arena.deinit(); self.* = undefined;}
/// Walks from the top, once the corpus has been read; or again once the/// walk under way ends, to keep what came above where it started.pub fn start(self: *Harvest) void { if (!self.corpus.ready) return; if (self.walking) { self.again = true; return; } _ = self.page_arena.reset(.retain_capacity); self.walking = true; self.next = ""; self.page = &.{}; self.wanted = &.{}; self.kept = 0; self.failures = 0; self.later(0);}
/// Stops until `start`; an answer on its way asks nothing more.pub fn stop(self: *Harvest) void { if (self.timer) |t| siv.timerCancel(t); self.timer = null; self.walking = false; self.again = false;}
fn later(self: *Harvest, 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: *Harvest = @ptrCast(@alignCast(user_data)); self.timer = null; if (!self.walking) return; if (self.host.busy(self.host.context)) return self.later(self.timing.busy_ms); self.step() catch |err| self.failed(err);}
/// Reads the first thread wanted, else lists the next page.fn step(self: *Harvest) !void { // A thread read whole can run to megabytes; most are far less. _ = self.request_arena.reset(.{ .retain_with_limit = 1 << 20 }); const a = self.request_arena.allocator(); if (self.wanted.len == 0) return self.gmail.listMessages(a, self.next.?, self, onPage); const w = self.wanted[0]; if (w.messages == 1) return self.gmail.readMessage(a, w.first, self, onRead); try self.gmail.readThread(a, w.thread, self, onRead);}
fn onPage(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *Harvest = @ptrCast(@alignCast(context)); if (!self.walking) return; self.takePage(response) catch |err| self.failed(err);}
/// Notes the page's threads with messages not kept, and the next page.fn takePage(self: *Harvest, response: Http.Error!Http.Response) !void { const body = try Gmail.check("GET /messages", self.request_arena.allocator(), response); // The ids are the body's, so it goes where they are kept; `next` is // kept until then, for a failure to try again. _ = self.page_arena.reset(.retain_capacity); const a = self.page_arena.allocator(); const page = try Gmail.messageIds(a, try a.dupe(u8, body)); var threads: std.array_hash_map.String(Want) = .empty; const ids = try a.alloc([]const u8, page.messages.len); for (page.messages, ids) |m, *id| { id.* = m.id; if (self.corpus.has(m.id)) continue; const t = try threads.getOrPut(a, m.threadId); if (!t.found_existing) t.value_ptr.* = .{ .thread = m.threadId, .first = m.id, .messages = 0 }; t.value_ptr.messages += 1; } self.page = ids; self.wanted = threads.values(); self.next = page.next; self.failures = 0; // Kept all of a page: what is below it was kept by an earlier walk. if (self.wanted.len == 0 and self.corpus.walked) return self.end(); self.goOn(page_units);}
fn onRead(context: *anyopaque, response: Http.Error!Http.Response) void { const self: *Harvest = @ptrCast(@alignCast(context)); if (!self.walking) return; self.keep(response) catch |err| self.failed(err);}
/// Keeps the messages read. One deleted since it was listed, or that/// cannot be made sense of, is passed over: it should not stop the walk.fn keep(self: *Harvest, response: Http.Error!Http.Response) !void { const w = self.wanted[0]; const a = self.request_arena.allocator(); const what = if (w.messages == 1) "GET /messages/…" else "GET /threads/… whole"; const read: []const Corpus.Record = if (Gmail.check(what, a, response)) |body| Gmail.readCorpus(a, body) catch |err| blk: { log.warn("{s} {s}: {t}", .{ what, w.thread, err }); break :blk &.{}; } else |err| if (err == error.NotFound) &.{} else return err; for (read) |r| { if (self.corpus.has(r.id)) continue; try self.corpus.add(r); self.kept += 1; if (self.kept % note_every == 0) log.info("kept {d} messages for search so far", .{self.kept}); } self.wanted = self.wanted[1..]; self.failures = 0; self.goOn(if (w.messages == 1) message_units else thread_units);}
/// Takes the next step once the quota spent is made up for, or ends the/// walk after the last page.fn goOn(self: *Harvest, units: u32) void { if (self.wanted.len == 0) { self.corpus.reach(self.page) catch |err| log.err("marking how far the walk got: {t}", .{err}); } if (self.wanted.len == 0 and self.next == null) { self.corpus.markWalked() catch |err| log.err("marking the walk's end: {t}", .{err}); return self.end(); } self.later(units * self.timing.unit_ms);}
fn end(self: *Harvest) void { if (self.kept > 0) log.info("kept {d} more messages for search, {d} in all", .{ self.kept, self.corpus.count() }); self.walking = false; if (!self.again) return; self.again = false; self.start();}
fn failed(self: *Harvest, err: anyerror) void { // An access token lasts an hour. The walk goes on once the host has // another; refused again, it waits as for any failure. if (err == error.Unauthorized and self.failures == 0) { log.info("{t}; trying again with a new access token", .{err}); self.host.refused(self.host.context); self.failures = 1; return self.later(0); } 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.page_arena.reset(.retain_capacity); self.next = ""; self.page = &.{}; self.wanted = &.{}; } self.failures +|= 1; self.later(wait);}
test "the walk reads whole what search lacks, a thread at once for more than one, and passes over one deleted" { var fake: Gmail.Fake = undefined; try fake.start(&routes); defer fake.stop(); var corpus: Corpus = undefined; corpus.open(testing.allocator, testing.io, fake.app, null, .{ .context = &corpus, .call = ignore }); defer corpus.deinit(); // m2 is kept already. m6 is refused once, as when the access token // expires: the walk goes on as soon as there is another, not after // the back-off. var m2 = testRecord("m2", "t2"); m2.date = 1_730_000_000; try corpus.add(m2); fake.fail_path = "/messages/m6?"; fake.fail_count.store(1, .monotonic); var host: TestHost = .{ .busy = true }; var harvest: Harvest = .init(testing.allocator, fake.app, &fake.gmail, &corpus, host.host(), .{ .unit_ms = 0, .busy_ms = 5, .retry_ms = 60_000, }); defer harvest.deinit(); harvest.start();
// Nothing while the host is busy. for (0..10) |_| try siv.appDispatch(fake.app, 10); try testing.expectEqual(0, fake.requests.load(.monotonic));
host.busy = false; try finish(&fake, &harvest); try testing.expect(corpus.walked); try testing.expectEqual(1, host.refused); for ([_][]const u8{ "m1", "m2", "m3", "m4", "m6" }) |id| try testing.expect(corpus.has(id)); // m5 was deleted after it was listed. try testing.expect(!corpus.has("m5")); // The walk has gone down as far as the pages' messages, not t1's // older m0, which it read with its thread. try testing.expect(corpus.has("m0")); try testing.expectEqual(1_730_000_000, corpus.reached); // Page 1; t1's two messages in one read; m3 alone, as its thread's // other message is kept; page 2; m5; m6, which fails once. const asked = try fake.asked(); const paths = [_][]const u8{ list_path, "/threads/t1?", "/messages/m3?", list_path ++ "&pageToken=p2", "/messages/m5?", "/messages/m6?", "/messages/m6?", }; try testing.expectEqual(paths.len, asked.len); for (paths, asked) |p, seen| try testing.expect(std.mem.startsWith(u8, seen.target, p)); try testing.expectEqual(1, fake.most_out.load(.monotonic));
// Walked once, the next walk ends at the first page it has all of. harvest.start(); try finish(&fake, &harvest); try testing.expectEqual(paths.len + 1, fake.requests.load(.monotonic));}
test "a walk asked for while one is under way follows it from the top" { var fake: Gmail.Fake = undefined; try fake.start(&routes); defer fake.stop(); var corpus: Corpus = undefined; corpus.open(testing.allocator, testing.io, fake.app, null, .{ .context = &corpus, .call = ignore }); defer corpus.deinit(); var host: TestHost = .{ .busy = false }; var harvest: Harvest = .init(testing.allocator, fake.app, &fake.gmail, &corpus, host.host(), .{ .unit_ms = 0, .busy_ms = 5, .retry_ms = 10, }); defer harvest.deinit(); harvest.start(); harvest.start(); try finish(&fake, &harvest); // The first walk goes all the way down: two pages and four reads. The // second ends at the first page. try testing.expect(corpus.walked); try testing.expectEqual(7, fake.requests.load(.monotonic));}
/// Runs the event loop until the walk ends.fn finish(fake: *Gmail.Fake, harvest: *const Harvest) !void { var waited: u32 = 0; while (harvest.walking and waited < 500) : (waited += 1) try siv.appDispatch(fake.app, 10); try testing.expect(!harvest.walking);}
fn ignore(_: *anyopaque) void {}
const TestHost = struct { busy: bool, /// How many times the walk asked for a new access token. refused: u32 = 0,
fn host(self: *TestHost) Host { return .{ .context = self, .busy = isBusy, .refused = onRefused }; }
fn onRefused(context: *anyopaque) void { const self: *TestHost = @ptrCast(@alignCast(context)); self.refused += 1; }
fn isBusy(context: *anyopaque) bool { const self: *TestHost = @ptrCast(@alignCast(context)); return self.busy; }};
fn testRecord(id: []const u8, thread: []const u8) Corpus.Record { return .{ .id = id, .thread = thread, .message_id = "", .subject = "", .list_id = "", .delivered_to = "", .date = 0, .size = 0, .from = .{ .name = "", .address = "" }, .to = &.{}, .cc = &.{}, .bcc = &.{}, .attachments = &.{}, .text = "", .quoted = "", };}
const list_path = "/messages?maxResults=500&q=-in%3Adrafts&fields=messages(id,threadId),nextPageToken";
/// A message as Gmail gives it whole, from Ana.fn message(comptime id: []const u8, comptime thread: []const u8) []const u8 { return messageAt(id, thread, "1730000000000");}
fn messageAt(comptime id: []const u8, comptime thread: []const u8, comptime ms: []const u8) []const u8 { return "{\"id\":\"" ++ id ++ "\",\"threadId\":\"" ++ thread ++ "\",\"labelIds\":[\"INBOX\"]," ++ "\"internalDate\":\"" ++ ms ++ "\",\"payload\":{\"headers\":" ++ "[{\"name\":\"From\",\"value\":\"ana@example.com\"}]}}";}
/// Page 1: t1's m1 and m4, and t2's m2 and m3; t1 holds m0 too, of years/// before, which no page lists. Page 2: m5, gone since, and m6.const routes = [_]Gmail.Fake.Route{ .ok(list_path ++ "&pageToken=p2", \\{"messages":[{"id":"m5","threadId":"t3"},{"id":"m6","threadId":"t4"}]} ), .ok(list_path, \\{"messages":[{"id":"m4","threadId":"t1"},{"id":"m3","threadId":"t2"},{"id":"m2","threadId":"t2"}, \\{"id":"m1","threadId":"t1"}],"nextPageToken":"p2"} ), .ok("/threads/t1?", "{\"messages\":[" ++ messageAt("m0", "t1", "1600000000000") ++ "," ++ message("m1", "t1") ++ "," ++ message("m4", "t1") ++ "]}"), .ok("/threads/t2?", "{\"messages\":[" ++ message("m2", "t2") ++ "," ++ message("m3", "t2") ++ "]}"), .ok("/messages/m3?", message("m3", "t2")), .ok("/messages/m6?", message("m6", "t4")), .{ .path = "/messages/m5?", .body = "{}", .status = .not_found },};