Skip to content

Set up logic for the final backup #2882

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dwwoelfel merged 3 commits into from
Aug 27, 2026
Merged

Set up logic for the final backup #2882

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9b94ecf
setup logic for the final backup
dwwoelfel Aug 27, 2026
e6d2baf
thread through expires-at
dwwoelfel Aug 27, 2026
dc1727e
default to expire
dwwoelfel Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 71 additions & 40 deletions server/src/instant/backup.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
[instant.model.schema :as schema]
[instant.model.webhook :as webhook-model]
[instant.storage.s3 :refer [s3-transfer-manager]]
[instant.sunset :as sunset]
[instant.util.async :as ua]
[instant.util.date :as date]
[instant.util.defrecord :refer [defrecord-once]]
Expand Down Expand Up @@ -117,18 +118,18 @@
conn
(uhsql/formatp insert-app-backup-q
{:id id
:app-id app-id
:isn isn
:backup-at backup-at
:storage-prefix storage-prefix
:files-size files-size
:db-size db-size
:uncompressed-size uncompressed-size
:description description
;; Can't be longer than 32 days because the storage rule
;; will automatically delete anything tagged with `expire`
;; after 32 days.
:expires-at expires-at}))))
:app-id app-id
:isn isn
:backup-at backup-at
:storage-prefix storage-prefix
:files-size files-size
:db-size db-size
:uncompressed-size uncompressed-size
:description description
;; Can't be longer than 32 days because the storage rule
;; will automatically delete anything tagged with `expire`
;; after 32 days.
:expires-at expires-at}))))

(def get-app-backups-by-app-id-q
(uhsql/preformat {:select [:id :app-id :isn :backup-at :files-size :db-size :uncompressed-size :description :expires-at]
Expand Down Expand Up @@ -360,7 +361,8 @@
isn
backup-at
description
expires-at]}
expires-at
expire-s3?]}
{:keys [streams app-id]}]
;; First, write the streams out
(->> (mapv (fn [^Stream stream]
Expand Down Expand Up @@ -407,7 +409,7 @@
backup-id)
:content-type "application/json"
:content-encoding "zstd"
:tags {:expire true}}
:tags {:expire expire-s3?}}
(AsyncRequestBody/fromBytes ba))

;; Once everything is uploaded to s3, we can insert the record into the db
Expand Down Expand Up @@ -491,7 +493,7 @@
(.writeEndObject gen)
(.writeRaw gen "\n")))

(defn start-stream [backup-id app-id etype]
(defn start-stream [backup-id app-id etype expire-s3?]
(let [pipe (Pipe. (long (* 256 1024))) ;; max in-flight bytes
sink (Okio/buffer (.sink pipe))
source (Okio/buffer (.source pipe))
Expand All @@ -514,7 +516,7 @@
etype)
:content-type "application/x-jsonlines"
:content-encoding "zstd"
:tags {:expire true}}
:tags {:expire expire-s3?}}
pipe-in))))
zstd-out (ZstdOutputStream. pipe-out compression-level)
;; Counter sits upstream of zstd so it sees the raw JSONL bytes
Expand Down Expand Up @@ -543,16 +545,16 @@
0
0)))

(defn ensure-stream [backup-id app-id ^HashMap streams etype]
(defn ensure-stream [backup-id app-id ^HashMap streams etype expire-s3?]
(if-let [stream (get streams etype)]
stream
(let [stream (start-stream backup-id app-id etype)]
(let [stream (start-stream backup-id app-id etype expire-s3?)]
(.put streams etype stream)
stream)))

(defn flush-entities [backup-id app-id streams entities]
(defn flush-entities [backup-id app-id streams entities expire-s3?]
(doseq [[etype ent] entities
:let [stream (ensure-stream backup-id app-id streams etype)]]
:let [stream (ensure-stream backup-id app-id streams etype expire-s3?)]]
(flush-ent stream ent)))

(defn snapshot-datasource
Expand Down Expand Up @@ -584,7 +586,8 @@
^LinkedBlockingQueue record-progress-queue
^LinkedBlockingQueue flush-streams-queue
on-done
done-signal]}]
done-signal
expire-s3?]}]
Comment thread
coderabbitai[bot] marked this conversation as resolved.
(loop [triple (.take triples-queue)
current-app-id (:app-id triple)
current-entity-id (:entity-id triple)
Expand All @@ -595,7 +598,7 @@
(cond (= triple done-signal)
(when current-app-id
(let [finished-promise (promise)]
(flush-entities process-id current-app-id streams entities)
(flush-entities process-id current-app-id streams entities expire-s3?)
(.put record-progress-queue {:app-id current-app-id
:triple-count triple-count
:finished-promise finished-promise})
Expand All @@ -606,7 +609,7 @@

(not= (:app-id triple) current-app-id)
(let [finished-promise (promise)]
(flush-entities process-id current-app-id streams entities)
(flush-entities process-id current-app-id streams entities expire-s3?)
(.put record-progress-queue {:app-id current-app-id
:triple-count triple-count
:finished-promise finished-promise})
Expand All @@ -622,7 +625,7 @@
triple-count))

(not= (:entity-id triple) current-entity-id)
(do (flush-entities process-id current-app-id streams entities)
(do (flush-entities process-id current-app-id streams entities expire-s3?)
(recur triple
current-app-id
(:entity-id triple)
Expand All @@ -640,7 +643,8 @@
(inc triple-count)))))
(on-done))

(defn process-with-snapshot []
(defn process-with-snapshot [{:keys [expire-s3?
expires-at]}]
(let [db-config (config/get-aurora-config)
process-id (random-uuid)
slot-name (str "backup_" (.replace (str process-id) "-" "_"))
Expand All @@ -654,8 +658,8 @@
:machine-id config/machine-id})
triples-queue (LinkedBlockingQueue. 50000)
record-progress-queue (LinkedBlockingQueue.)
flush-streams-queue (LinkedBlockingQueue. 100)
flush-stream-process-count 100
flush-streams-queue (LinkedBlockingQueue. 8)
flush-stream-process-count 8
done-signal ::done
process-state (atom nil)
abort (fn []
Expand Down Expand Up @@ -683,7 +687,8 @@
(.put record-progress-queue done-signal)
(dotimes [_ flush-stream-process-count]
(.put flush-streams-queue done-signal)))
:done-signal done-signal}))
:done-signal done-signal
:expire-s3? expire-s3?}))
flush-streams-processes (mapv (fn [_]
(ua/vfuture
(loop [item (.take flush-streams-queue)]
Expand All @@ -692,7 +697,9 @@
(complete-streams query-conn
{:backup-id process-id
:isn isn
:backup-at before-ts}
:backup-at before-ts
:expire-s3? expire-s3?
:expires-at expires-at}
item)
(deliver (:finished-promise item) true)
(catch Throwable t
Expand Down Expand Up @@ -798,7 +805,8 @@
expires-at
ensure-config?
on-triple
finished-promise]}]
finished-promise
expire-s3?]}]
(try
(let [triples-queue (LinkedBlockingQueue. 5000)
upload-progress-queue (LinkedBlockingQueue.)
Expand All @@ -824,7 +832,8 @@
:flush-streams-queue flush-streams-queue
:on-done (fn []
(.put flush-streams-queue done-signal))
:done-signal done-signal})
:done-signal done-signal
:expire-s3? expire-s3?})
(catch Throwable t
(deliver finished-promise t)
(.put flush-streams-queue done-signal)
Expand All @@ -835,7 +844,8 @@
:isn isn
:backup-at backup-at
:description description
:expires-at expires-at}
:expires-at expires-at
:expire-s3? expire-s3?}
;; This is a little awkward, since we'll only ever take 1 thing out of the queue,
;; but it allows us to use the same code as `process-with-copy`
item (.take flush-streams-queue)]
Expand Down Expand Up @@ -871,7 +881,7 @@

Returns {:triple-count <long> :app-backup <row>} where the row is the
inserted `app_backups` record. Throws if the backup fails."
[{:keys [app-id description expires-at on-triple]}]
[{:keys [app-id description expires-at on-triple expire-s3?]}]
(let [backup-id (random-uuid)]
(with-open [conn (wal/get-pg-copy-ready-conn (config/get-aurora-config))]
(.setAutoCommit conn false)
Expand All @@ -890,7 +900,8 @@
:expires-at expires-at
:ensure-config? true
:on-triple on-triple
:finished-promise finished-promise})
:finished-promise finished-promise
:expire-s3? expire-s3?})
(let [result @finished-promise]
(when (instance? Throwable result)
(throw result))
Expand Down Expand Up @@ -923,7 +934,9 @@
[{:keys [clone-pool
clone-lsn
^long process-count
backup-at]}]
backup-at
expire-s3?
expires-at]}]
(let [process-id (random-uuid)
isn (instant.isn/->ISN config/invalidator-slot-num clone-lsn)
_ (insert-backup-job! {:id process-id
Expand Down Expand Up @@ -969,7 +982,9 @@
:isn isn
:backup-at backup-at
:app-id (:app-id item)
:finished-promise (:finished-promise item)})
:finished-promise (:finished-promise item)
:expire-s3? expire-s3?
:expires-at expires-at})
(catch Throwable t
(tracer/with-new-trace-root
(tracer/record-exception-span! t {:name "backup/handle-app-error"}))
Expand Down Expand Up @@ -1024,7 +1039,8 @@
(defn process-with-clone
"Similar to process-with-copy, but creates a clone of the production database.
Allows us to process multiple apps concurrently"
[{:keys [source-cluster-id]}]
[{:keys [source-cluster-id expire-s3? expires-at]
:or {expire-s3? true}}]
(let [clone-config (clone/create-clone! {:instance-class "db.r8gd.xlarge"
:source-cluster-id source-cluster-id})]
(try
Expand All @@ -1037,7 +1053,9 @@
process (process-with-clone-pool {:clone-pool clone-pool
:process-count process-count
:clone-lsn clone-lsn
:backup-at backup-at})]
:backup-at backup-at
:expire-s3? expire-s3?
:expires-at expires-at})]
((:wait-for-finish process)))

(finally
Expand All @@ -1051,7 +1069,8 @@
(def -process (process-with-clone-pool {:clone-pool (aurora/conn-pool :write)
:clone-lsn (:lsn (sql/select-one (aurora/conn-pool :read) ["select pg_current_wal_lsn() as lsn"]))
:backup-at (Instant/now)
:process-count 10})))
:process-count 10
:expire-s3? true})))

;; Scheduled nightly backup
;;
Expand All @@ -1078,7 +1097,19 @@
(tracer/with-span! {:name "backup/daily-backup"
:attributes {:date date-str}}
(try
(process-with-clone {:source-cluster-id (clone/default-source-cluster-id)})
(let [final-backup-status (flags/flag :final-backup-status)
final-snapshot? (= "pending" final-backup-status)
backup-config (merge {:expire-s3? (not final-snapshot?)}
(when final-snapshot?
{:expires-at (.plus (Instant/now) (Duration/ofDays (+ 365 32)))}))]
(when-not (= "completed" final-backup-status)
(if (flags/toggled? :backup-with-clone true)
(process-with-clone (assoc backup-config
:source-cluster-id (clone/default-source-cluster-id)))
(let [process (process-with-snapshot backup-config)]
((:wait-for-finish process)))))
(when final-snapshot?
(sunset/update-flag! "final-backup-status" "completed")))
Comment thread
coderabbitai[bot] marked this conversation as resolved.
(swap! current-backup assoc
:status :completed
:finished-at (Instant/now))
Expand Down
1 change: 1 addition & 0 deletions server/src/instant/db/app_backup_jobs.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -338,6 +338,7 @@
(backup/backup-app-on-primary! {:app-id app_id
:description (or description "On-demand backup")
:expires-at expires-at
:expire-s3? true
:on-triple (fn [] (.incrementAndGet copied))})
(finally
(deliver unwound true))))]
Expand Down
6 changes: 4 additions & 2 deletions server/src/instant/sunset.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@
;; ----------
;; Updating the flag

(defn- update-flag!
(defn update-flag!
"Upserts a `flags` entity in the config app. Relies on `flags.setting`
being unique so a lookup ref works as an upsert."
[setting value]
Expand Down Expand Up @@ -103,7 +103,9 @@
[{:message (str "Stage must be one of: "
(string/join ", " (map name flags/sunset-stages))
".")}]))
(update-flag! "sunset-stage" (name stage)))
(update-flag! "sunset-stage" (name stage))
(when (= stage :read-only)
(update-flag! "final-backup-status" "pending")))
Comment thread
coderabbitai[bot] marked this conversation as resolved.

(defn state
"Current sunset state as seen by this machine's flag subscription."
Expand Down
Loading