Источник (лёгкая очередь) отдаёт данные мелкими порциями по десятку записей,
а приёмник (тяжёлая СУБД) эффективен только на крупных батчах в тысячи
записей. Нужно перекачивать данные из источника в приёмник.
Источник бесконечен. Каждая порция приходит с cookie — меткой позиции.
После перезапуска источник продолжает с последней подтверждённой позиции,
поэтому подтверждать (Ack) можно только то, что уже надёжно сохранено, и
строго в порядке получения.
const capPerCall = 4999
type Source interface {
// Fetch возвращает очередную порцию (не больше capPerCall записей,
// возможно пустую), cookie для подтверждения и ошибку.
Fetch() (items []any, cookie int, err error)
Ack(cookie int) error
}
type Sink interface {
// Save принимает не больше capPerCall записей за вызов.
Save(items []any) error
}
// Flow перекачивает данные, пока не случится ошибка, и возвращает её.
func Flow(src Source, sink Sink) error
Требования
- Записи попадают в приёмник в том порядке, в каком пришли из источника,
без потерь и дублей. Порция не разрезается между батчами.
- Каждый
Save получает непустой батч не больше capPerCall записей.
Батчи крупные: батч отправляется, только когда очередная непустая
порция в него уже не помещается.
- Каждый полученный cookie подтверждается
Ack ровно один раз, в порядке
получения и только после того, как все записи этой порции (и всех
предыдущих) сохранены.
- Любая ошибка
Fetch, Save или Ack завершает Flow и возвращается
(можно обернуть, errors.Is должен её находить). Cookie несохранённых
порций не подтверждаются. Сохранять ли остаток буфера при ошибке Fetch —
на твоё усмотрение.
- Код проходит
go test -race.
Пример
// порции по 30 записей: Save получает 166 порций = 4980 записей
// (167-я уже не влезает в 4999), потом Ack для 166 cookie по порядку
err := Flow(queue, db)
Fetch и Save медленные, и ждать Save, чтобы начать копить следующий
батч, расточительно.
func FlowParallel(src Source, sink Sink) error
- Пока идёт
Save одного батча, следующий батч уже набирается через Fetch.
- Все требования
Flow сохраняются: порядок записей, крупные батчи, Ack
строго по порядку и после сохранения.
Save не вызывается конкурентно сам с собой, Fetch — сам с собой
(Fetch и Ack друг с другом — можно).- Память ограничена: в ожидании
Save не больше пары батчей, даже если
Fetch намного быстрее.
- После возврата
FlowParallel горутин не остаётся.
Устно: что получит приёмник после падения процесса между Save и Ack
и как сделать запись идемпотентной? Почему достаточно было бы подтверждать
только последний cookie батча?