graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Chunk Oban.insert_all to avoid PG parameter limit

Browse source 
Weather job fan-out can exceed PostgreSQL's 65535 parameter limit
when many QSOs produce thousands of jobs in a single batch. Chunk
insert_all into batches of 1000 jobs.
This commit is contained in:
Graham McIntire 2026-03-30 13:31:48 -05:00
parent 7f97c72f98
commit 2006e9e052
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
1 changed files with 12 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
lib/microwaveprop/workers/qso_weather_enqueue_worker.ex
View file

@ -12,6 +12,8 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
@asos_radius_km 150 @asos_radius_km 150
@sounding_radius_km 300 @sounding_radius_km 300
# Oban jobs have ~15 params each; PG limit is 65535 params per query
@insert_batch_size 1000
@impl Oban.Worker @impl Oban.Worker
def perform(%Oban.Job{}) do def perform(%Oban.Job{}) do
@ -34,7 +36,7 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
jobs = build_weather_jobs(qsos) jobs = build_weather_jobs(qsos)
if jobs != [] do if jobs != [] do
Oban.insert_all(jobs) insert_all_chunked(jobs)
end end
qso_ids = Enum.map(qsos, & &1.id) qso_ids = Enum.map(qsos, & &1.id)
@ -53,7 +55,7 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
jobs = build_hrrr_jobs(qsos) jobs = build_hrrr_jobs(qsos)
if jobs != [] do if jobs != [] do
Oban.insert_all(jobs) insert_all_chunked(jobs)
end end
qso_ids = Enum.map(qsos, & &1.id) qso_ids = Enum.map(qsos, & &1.id)
@ -72,7 +74,7 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
jobs = build_terrain_jobs(qsos) jobs = build_terrain_jobs(qsos)
if jobs != [] do if jobs != [] do
Oban.insert_all(jobs) insert_all_chunked(jobs)
end end
qso_ids = Enum.map(qsos, & &1.id) qso_ids = Enum.map(qsos, & &1.id)
@ -128,7 +130,7 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
jobs = build_iemre_jobs(qsos) jobs = build_iemre_jobs(qsos)
if jobs != [] do if jobs != [] do
Oban.insert_all(jobs) insert_all_chunked(jobs)
end end
qso_ids = Enum.map(qsos, & &1.id) qso_ids = Enum.map(qsos, & &1.id)
@ -213,4 +215,10 @@ defmodule Microwaveprop.Workers.QsoWeatherEnqueueWorker do
}) })
end end
end end
defp insert_all_chunked(jobs) do
jobs
|> Enum.chunk_every(@insert_batch_size)
|> Enum.each(&Oban.insert_all/1)
end
end end