| 40 | 40 | | const KEPT_DAYS: u64 = 14; |
| 41 | 41 | | /// How many due retries one sweep makes. |
| 42 | 42 | | const SWEEP: u32 = 50; |
| 43 | + | /// The hourly cron that forgets old deliveries (wrangler.jsonc); every |
| 44 | + | /// other minute only retries. |
| 45 | + | const PURGE_CRON: &str = "37 * * * *"; |
| 46 | + | /// How many old deliveries one statement deletes, and how many statements |
| 47 | + | /// one purge makes: a busy hour's worth, in bites D1 answers quickly. |
| 48 | + | const PURGE_BATCH: u32 = 1_000; |
| 49 | + | const PURGE_BATCHES: u32 = 50; |
| 43 | 50 | | |
| 44 | 51 | | #[derive(Deserialize)] |
| 45 | 52 | | struct HookRow { |
| ⋯ |
| 661 | 668 | | Ok(()) |
| 662 | 669 | | } |
| 663 | 670 | | |
| 664 | | − | /// Tries again what is due, and forgets what is old. |
| 671 | + | /// Tries again what is due. |
| 665 | 672 | | async fn sweep(&self) -> Result<()> { |
| 666 | 673 | | let now = now_ms(); |
| 667 | 674 | | let due = self |
| ⋯ |
| 690 | 697 | | } |
| 691 | 698 | | } |
| 692 | 699 | | } |
| 693 | | − | self.db |
| 694 | | − | .prepare("DELETE FROM deliveries WHERE created_at < ?") |
| 695 | | − | .bind(&[rfc3339(now.saturating_sub(KEPT_DAYS * 24 * 60 * 60 * 1000)).into()])? |
| 696 | | − | .run() |
| 697 | | − | .await?; |
| 698 | 700 | | Ok(()) |
| 699 | 701 | | } |
| 702 | + | |
| 703 | + | /// Forgets deliveries older than [`KEPT_DAYS`], oldest first, a batch |
| 704 | + | /// at a time (`deliveries_by_time`, migration 0002). What one purge |
| 705 | + | /// leaves, the next hour's takes. Returns how many it deleted. |
| 706 | + | async fn purge(&self) -> Result<u32> { |
| 707 | + | let before = rfc3339(now_ms().saturating_sub(KEPT_DAYS * 24 * 60 * 60 * 1000)); |
| 708 | + | let mut deleted = 0; |
| 709 | + | for _ in 0..PURGE_BATCHES { |
| 710 | + | let result = self |
| 711 | + | .db |
| 712 | + | .prepare(PURGE_SQL) |
| 713 | + | .bind(&[before.as_str().into(), PURGE_BATCH.into()])? |
| 714 | + | .run() |
| 715 | + | .await?; |
| 716 | + | let changed = result.meta()?.and_then(|meta| meta.changes).unwrap_or(0) as u32; |
| 717 | + | deleted += changed; |
| 718 | + | if changed < PURGE_BATCH { |
| 719 | + | break; |
| 720 | + | } |
| 721 | + | } |
| 722 | + | Ok(deleted) |
| 723 | + | } |
| 700 | 724 | | } |
| 701 | 725 | | |
| 702 | 726 | | /// One HTTPS POST of a delivery's payload, signed, given ten seconds. |
| ⋯ |
| 810 | 834 | | Ok(()) |
| 811 | 835 | | } |
| 812 | 836 | | |
| 813 | | − | /// Every minute: retries that are due, and deliveries old enough to forget. |
| 837 | + | /// One batch of old deliveries, by time. |
| 838 | + | const PURGE_SQL: &str = "DELETE FROM deliveries WHERE rowid IN (SELECT rowid FROM deliveries WHERE created_at < ? ORDER BY created_at LIMIT ?)"; |
| 839 | + | |
| 840 | + | /// Every minute: retries that are due. Once an hour, at [`PURGE_CRON`]: |
| 841 | + | /// deliveries old enough to forget. |
| 814 | 842 | | #[event(scheduled)] |
| 815 | | − | async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) { |
| 816 | | − | match Webhooks::new(&env) { |
| 817 | | − | Ok(service) => { |
| 818 | | − | if let Err(error) = service.sweep().await { |
| 819 | | − | worker::console_error!("webhooks: the sweep failed: {error}"); |
| 820 | | − | } |
| 843 | + | async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) { |
| 844 | + | let service = match Webhooks::new(&env) { |
| 845 | + | Ok(service) => service, |
| 846 | + | Err(error) => { |
| 847 | + | worker::console_error!("webhooks: could not start: {error}"); |
| 848 | + | return; |
| 849 | + | } |
| 850 | + | }; |
| 851 | + | if event.cron() == PURGE_CRON { |
| 852 | + | match service.purge().await { |
| 853 | + | Ok(0) => {} |
| 854 | + | Ok(deleted) => worker::console_log!("webhooks: forgot {deleted} old deliveries"), |
| 855 | + | Err(error) => worker::console_error!("webhooks: the purge failed: {error}"), |
| 821 | 856 | | } |
| 822 | | − | Err(error) => worker::console_error!("webhooks: could not start: {error}"), |
| 857 | + | return; |
| 858 | + | } |
| 859 | + | if let Err(error) = service.sweep().await { |
| 860 | + | worker::console_error!("webhooks: the sweep failed: {error}"); |
| 861 | + | } |
| 862 | + | } |
| 863 | + | |
| 864 | + | #[cfg(test)] |
| 865 | + | mod tests { |
| 866 | + | use super::*; |
| 867 | + | |
| 868 | + | #[test] |
| 869 | + | fn the_purge_has_a_cron_of_its_own() { |
| 870 | + | let wrangler = include_str!("../wrangler.jsonc"); |
| 871 | + | assert!(wrangler.contains(&format!("\"{PURGE_CRON}\"")), "{PURGE_CRON} is not among the crons"); |
| 872 | + | assert_ne!(PURGE_CRON, "* * * * *"); |
| 873 | + | } |
| 874 | + | |
| 875 | + | #[test] |
| 876 | + | fn the_purge_deletes_in_batches_by_time() { |
| 877 | + | assert!(PURGE_SQL.contains("WHERE created_at < ? ORDER BY created_at LIMIT ?")); |
| 878 | + | assert!(PURGE_BATCH * PURGE_BATCHES >= 10_000); |
| 823 | 879 | | } |
| 824 | 880 | | } |
| 825 | 881 | | |
