Change Streams: события реального времени

Подписка строится одним методом с тремя владельцами: client.Watch ловит события всего кластера, db.Watch — всех коллекций одной базы, coll.Watch — единственной коллекции. Область видимости — единственное различие; сигнатура во всех трёх случаях одна и та же: Watch(ctx, pipeline, opts...) и *ChangeStream на выходе. В v2 конвейер описывается значением типа any: подойдёт mongo.Pipeline из bson.D, знакомый по агрегациям, или пустой конвейер для всех событий без фильтра. Требования к окружению жёсткие, и о них стоит помнить до первого запуска: события собираются из oplog, поэтому нужен replica set — на одиночном сервере Watch вернёт ошибку сразу же, при создании потока. Вторая оговорка из документации драйвера: клиент должен работать с read concern majority либо вообще без заданного read concern, иначе сервер откажется создать поток. Сам ChangeStream — обёртка над курсором, и механика чтения из седьмого урока переносится один в один: Next, Decode, Err и Close — те же имена, те же правила завершения. Разница в характере курсора: обычный кончается, когда документы кончились, а поток живёт бесконечно и ждёт новых событий.

Событие — обычный BSON-документ, и главные его поля известны заранее. operationType называет операцию: insert, update, replace, delete или invalidate — пять типов, на которые стоит закладываться обработчику. documentKey несёт _id затронутого документа — тот самый ObjectID из двенадцатого урока, по которому событие всегда привязывается к записи, даже когда самого документа уже нет. ns сообщает базу и коллекцию, clusterTime — момент операции. Для insert и replace в событии лежит fullDocument — полный документ после операции; с update иначе: по умолчанию приезжает только updateDescription, список изменённых полей из $set, а полного документа нет. Это экономно для больших записей, но ломает обработчик, рассчитанный на готовый объект. Плечо, которым документ подтягивают целиком, — опция SetFullDocument(options.UpdateLookup) на конструкторе options.ChangeStream(): в каждое событие update вставляется текущее состояние документа. Нюанс, о котором помнят в проде: UpdateLookup читает документ на момент получения события, а не на момент изменения, — между этими мгновениями чужие правки уже могли пролечь, и подписчик видит чуть более свежую версию, чем та, что вызвала событие. Декодировать событие удобнее структурой с bson-тегами из пятого урока: поля operationType, documentKey и fullDocument описываются в типах один раз и живут в коде как обычные объекты. Полный список полей события длиннее — в событиях транзакций появляются lsid и txnNumber, — но перечисленных пяти хватает для подавляющего большинства подписчиков, а остальные стоит открывать только когда базовый обработчик уже работает.

Первым аргументом Watch принимает конвейер агрегаций — тот же язык этапов, что в пятнадцатом уроке, только применённый к потоку событий, а не к коллекции. Чаще всего хватает одного $match: оставить вставки или события определённого типа, и сервер отдаёт поток уже отфильтрованным — мусор до драйвера не доезжает вовсе. Конвейер на событиях компактен: bson.D с ключом operationType и $in со списком типов, тем же синтаксисом, что в обычных фильтрах. Доступны и другие этапы — $addFields и $project обогащают и сужают событие, — но увлекаться не стоит: тяжёлый конвейер живёт на сервере всё время подписки и ест ресурсы кластера, пока подписчик подключён. Фильтр при этом уместен почти всегда: подписчик, отбрасывающий девять событий из десяти уже в Go, тратит сеть и процессор на события, которые сервер мог бы не отправлять, — тот же $match на входе Watch снимает проблему бесплатно. Этап $project дополнительно сужает событие до нужных полей: обработчику, которому важны только _id и тип операции, не нужен весь документ с десятками полей. На клиенте, базе и коллекции конвейер фильтрует одинаково — разница только в объёме потока, который дойдёт до приложения.

Change Streams живут в выделенной горутине, и жизненный цикл у неё простой, как у любого фонового воркера. Next(ctx) блокируется, пока не придёт событие, не случится ошибка или не истечёт контекст: при отмене ctx метод вернёт false, а cs.Err() отдаст контекстную ошибку — поток умирает вместе со своим контекстом, и это штатный способ остановки: отменили контекст сервиса, горутина вышла, defer cs.Close закрыл поток. Для неблокирующего чтения существует TryNext: он возвращается сразу, событие есть или нет; если false, а Err() пуст, событий просто не было — цикл ждёт и зовёт снова. После false от Next дальнейшие вызовы тоже вернут false: поток в ошибочном состоянии, читайте Err и решайте, пересоздавать Watch или отдавать ошибку наверх. Внутри драйвера поток сам переживает мелкие неприятности — временный обрыв сети и смену узла реплики: резюм по токену встроен, и цикл Next таких качелей обычно даже не замечает. Наружу прорывается то, что драйвер вылечить не может: invalidate, потерянная точка продолжения или длительная смерть кластера. Горутине при этом нужен собственный контекст, производный от контекста сервиса: поделите время жизни так, чтобы отмена сервиса гарантированно гасила подписку, а не оставляла её висеть в фоне с открытым курсором.

Каждое событие потока несёт резюм-токен: cs.ResumeToken() возвращает bson.Raw с меткой последнего прочитанного события. Токен — точка продолжения, и у драйвера три опции на этот счёт. SetResumeAfter(token) начинает чтение с события после сохранённого токена — канонический способ продолжить с места остановки. SetStartAfter(point) работает похоже, но стартует после указанной точки без чтения её самой, а SetStartAtOperationTime привязывает начало потока к моменту времени через *bson.Timestamp. Драйвер пользуется токенами непрерывно и без вашей помощи: временная ошибка, переключение на другой узел реплики — поток резюмируется внутри, и цикл Next этого не заметит. Ручное сохранение нужно ровно для одного сценария — рестарта приложения: при каждом событии пишите токен в базу или файл, при старте читайте его и передавайте в SetResumeAfter, тогда события за время простоя доедут из oplog-истории. У истории есть глубина: oplog конечен, и если приложение лежало дольше его окна, сервер ответит ошибкой о том, что точка продолжения вышла из oplog, — лечится это полной пересверкой данных, а не доигрыванием событий. Порядок действий при сохранении тоже важен: сначала обработать событие, потом записать токен — при падении посередине событие обработается повторно, и подписчик обязан выдерживать такой повтор без порчи данных.