MongoDB Document Validation Drift Repair Guard with Kestra

id: mongodb-document-validation-drift-repair-guard
namespace: company.team

description: |
  Find the documents in a MongoDB collection that do not match the collection's own
  `$jsonSchema` validator. A validator added with `collMod` applies to new writes only, so every
  document written before it stays in place and breaks the rule nobody checked. This flow reads
  the collection with `$nor` on the schema, splits the violators into the ones a default can
  repair and the ones it must refuse to guess at, backfills the missing required field behind a
  safe mode gate and an approval, then reads the collection again and fails the run if any
  repairable violator is still there.

inputs:
  - id: connection_uri
    type: STRING
    displayName: MongoDB connection string
    description: >-
      Leave empty in production so the flow reads the MONGODB_URI secret. Set it
      to a plain connection string only for a local test run.
    defaults: ""

  - id: database
    type: STRING
    displayName: Database
    defaults: appdata

  - id: collection
    type: STRING
    displayName: Collection
    defaults: orders

  - id: json_schema
    type: STRING
    displayName: The collection validator schema
    description: >-
      The `$jsonSchema` object the collection is validated against, as JSON.
      Read it from `db.getCollectionInfos({name:
      'orders'})[0].options.validator`. Passing it as an input keeps the flow
      reusable across collections.
    defaults: '{"bsonType":"object","required":["order_id","status"],"properties":{"order_id":{"bsonType":"string"},"status":{"bsonType":"string"}}}'

  - id: required_field
    type: STRING
    displayName: Required field to backfill
    description: >-
      The one missing required field this run repairs. It must be named in the
      schema `required` list. Documents that violate the schema in any other way
      are reported untouched.
    defaults: status

  - id: default_value
    type: STRING
    displayName: Default for the missing field
    description: >-
      A JSON literal written into the missing field. Quote a string value.
      Numbers and booleans go in bare. The flow refuses the run when its type
      contradicts the `bsonType` the schema declares for the field.
    defaults: '"unknown"'

  - id: repair_limit
    type: INT
    displayName: Documents repaired per run
    description: >-
      Ceiling on how many documents one run backfills. The counts in the report
      are exact and server-side. Only the repair list is capped, so a very large
      drift is cleared over several runs.
    defaults: 1000

  - id: batch_size
    type: INT
    displayName: Documents per write batch
    description: >-
      The repair writes in batches of this many document ids, one batch at a
      time, so a large backfill never becomes one long running write.
    defaults: 250

  - id: sample_limit
    type: INT
    displayName: Documents named per report
    description: Ceiling on how many refused documents the report names and on each
      re-check read.
    defaults: 25

  - id: approval_timeout
    type: STRING
    displayName: Approval timeout
    description: ISO-8601 duration to wait for an answer before the pause expires.
    defaults: "PT2H"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow the backfill. Still needs dry_run false and an approval.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, before any document is
      written.
    defaults: true

variables:
  input_check: >-
    {{ {'s': fromJson(inputs.json_schema), 'f': inputs.required_field, 'd':
    inputs.default_value} | jq('.f as $f | .s as $s | .d as $d | (($s.required)
    // []) as $req | ((($s.properties) // {})[$f].bsonType) as $bt | (($d |
    [fromjson]?) // []) as $w | ($w[0]) as $v | (if ($bt | type) != "string"
    then null elif $bt == "string" then "string" elif ($bt == "int" or $bt ==
    "long" or $bt == "double" or $bt == "decimal" or $bt == "number") then
    "number" elif $bt == "bool" then "boolean" elif $bt == "object" then
    "object" elif $bt == "array" then "array" else null end) as $want | if ($req
    | index($f)) == null then ("required_field " + $f + " is not one of the
    schema required fields " + ($req | tojson)) elif ($w | length) == 0 then
    "default_value is not a JSON literal. Quote a string value. Numbers and
    booleans go in bare." elif ($want != null and ($v | type) != $want) then
    ("default_value is a " + ($v | type) + " but the schema declares " + $f + "
    as bsonType " + $bt) else "ok" end') | first }}

  detect_pipeline: >-
    {{ {'s': fromJson(inputs.json_schema), 'f': inputs.required_field, 'rl':
    inputs.repair_limit, 'sl': inputs.sample_limit} | jq('.f as $f | .s as $s |
    ($s | .required = ((($s.required) // []) | map(select(. != $f)))) as $sm |
    [{"$match": {"$nor": [{"$jsonSchema": $s}]}}, {"$facet": {"ids": [{"$match":
    {"$and": [{($f): {"$exists": false}}, {"$jsonSchema": $sm}]}}, {"$project":
    {"oid": {"$toString": "$_id"}}}, {"$limit": .rl}], "rcount": [{"$match":
    {"$and": [{($f): {"$exists": false}}, {"$jsonSchema": $sm}]}}, {"$count":
    "n"}], "other": [{"$match": {"$nor": [{"$and": [{($f): {"$exists": false}},
    {"$jsonSchema": $sm}]}]}}, {"$project": {"_id": 0, "document_id":
    {"$toString": "$_id"}, "value_type": {"$type": ("$" + $f)},
    "missing_required": {"$setDifference": [(($s.required) // []), {"$map":
    {"input": {"$objectToArray": "$$ROOT"}, "as": "kv", "in": "$$kv.k"}}]}}},
    {"$limit": .sl}], "total": [{"$count": "n"}]}}, {"$project": {"repair_ids":
    {"$map": {"input": "$ids", "as": "x", "in": "$$x.oid"}}, "repairable":
    {"$ifNull": [{"$arrayElemAt": ["$rcount.n", 0]}, 0]}, "violators":
    {"$ifNull": [{"$arrayElemAt": ["$total.n", 0]}, 0]}, "refused": "$other"}},
    {"$set": {"targeted": {"$size": "$repair_ids"}, "unrepairable":
    {"$subtract": ["$violators", "$repairable"]}}}]') | first | toJson }}

  set_document: >-
    {{ {'f': inputs.required_field, 'd': inputs.default_value} | jq('{"$set":
    {(.f): (.d | fromjson)}}') | first | toJson }}

  repair_batches: >-
    {{ {'ids': outputs.detect.rows[0].repair_ids, 'n': inputs.batch_size} |
    jq('.n as $n | [.ids | to_entries | group_by(.key / $n | floor) | .[] |
    map({"$oid": .value}) | tojson]') | first | toJson }}

  repairable_filter: >-
    {{ {'s': fromJson(inputs.json_schema), 'f': inputs.required_field} | jq('.f
    as $f | .s as $s | ($s | .required = ((($s.required) // []) | map(select(.
    != $f)))) as $sm | {"$and": [{"$nor": [{"$jsonSchema": $s}]}, {($f):
    {"$exists": false}}, {"$jsonSchema": $sm}]}') | first | toJson }}

  verify_pipeline: >-
    {{ {'s': fromJson(inputs.json_schema), 'f': inputs.required_field, 'ids':
    outputs.detect.rows[0].repair_ids} | jq('.f as $f | .s as $s | (.ids |
    map({"$oid": .})) as $oids | [{"$facet": {"valid": [{"$match": {"$and":
    [{"_id": {"$in": $oids}}, {"$jsonSchema": $s}]}}, {"$count": "n"}],
    "missing": [{"$match": {"$and": [{"_id": {"$in": $oids}}, {($f): {"$exists":
    false}}]}}, {"$count": "n"}], "compliant": [{"$match": {"$jsonSchema": $s}},
    {"$count": "n"}]}}, {"$project": {"targeted_valid": {"$ifNull":
    [{"$arrayElemAt": ["$valid.n", 0]}, 0]}, "targeted_missing": {"$ifNull":
    [{"$arrayElemAt": ["$missing.n", 0]}, 0]}, "compliant": {"$ifNull":
    [{"$arrayElemAt": ["$compliant.n", 0]}, 0]}}}]') | first | toJson }}

tasks:
  - id: check_inputs
    type: io.kestra.plugin.core.flow.If
    description: >-
      Refuse a field the schema does not require. Refuse a default whose JSON
      type contradicts the bsonType the schema declares for that field. Both are
      checked before a single read, because a default of the wrong type is how a
      repair run turns one broken rule into two.
    condition: "{{ render(vars.input_check) != 'ok' }}"
    then:
      - id: bad_inputs
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          Input check failed. {{ render(vars.input_check) }}. Nothing was read.
          Nothing was written.

  - id: detect
    type: io.kestra.plugin.mongodb.Aggregate
    description: >-
      Read-only. Matches every document the validator would now reject with
      `$nor` on the schema, then splits them. A violator whose only problem is
      the missing required field is repairable, which is tested by matching it
      against the same schema with that field dropped from `required`.
      Everything else is refused with its BSON value type and its missing
      required keys named. Counts are exact and server-side. Only the repair id
      list is capped.
    connection:
      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
        secret('MONGODB_URI') }}"
    database: "{{ inputs.database }}"
    collection: "{{ inputs.collection }}"
    pipeline: "{{ render(vars.detect_pipeline) }}"
    store: FETCH

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: Log the counts. Task outputs are not in the execution JSON, so the
      numbers the gates use are written here.
    message: >-
      Checked {{ inputs.database }}.{{ inputs.collection }} against its
      validator schema. Documents the validator would reject: {{
      outputs.detect.rows[0].violators ?? 0 }}. Repairable by backfilling {{
      inputs.required_field }}: {{ outputs.detect.rows[0].repairable ?? 0 }}.
      Targeted this run: {{ outputs.detect.rows[0].targeted ?? 0 }}
      (repair_limit {{ inputs.repair_limit }}). Refused because a default cannot
      fix them: {{ outputs.detect.rows[0].unrepairable ?? 0 }}. Refused
      documents: {{ outputs.detect.rows[0].refused ?? 'none' }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: Any violator means alert and consider the backfill. A collection
      that matches its own validator just logs and stops.
    condition: "{{ (outputs.detect.rows[0].violators ?? 0) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': ('MongoDB validation drift on ' ~ inputs.database ~ '.' ~ inputs.collection ~ ': ' ~ (outputs.detect.rows[0].violators ?? 0) ~ ' document(s) do not match the collection validator. ' ~ (outputs.detect.rows[0].repairable ?? 0) ~ ' can be repaired by backfilling ' ~ inputs.required_field ~ '. ' ~ (outputs.detect.rows[0].unrepairable ?? 0) ~ ' need a human because a default cannot fix them. Execution ' ~ execution.id ~ '.')} | toJson }}

      - id: planned
        type: io.kestra.plugin.core.log.Log
        description: What a remediation run would do, printed on every run including a
          dry one.
        message: >-
          A repair run would set {{ inputs.required_field }} to {{
          inputs.default_value }} on {{ outputs.detect.rows[0].targeted ?? 0 }}
          document(s) in batches of {{ inputs.batch_size }}. It would leave the
          {{ outputs.detect.rows[0].unrepairable ?? 0 }} refused document(s)
          exactly as they are. dry_run={{ inputs.dry_run }} auto_remediate={{
          inputs.auto_remediate }}. A write runs only when auto_remediate is
          true and dry_run is false and a reviewer approves the pause.

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        description: The backfill opens only when auto_remediate is true and dry_run is
          false.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: hold
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              A backfill writes a value somebody chose into documents a human
              never looked at, so a reviewer answers first. With no answer
              nothing is written and the findings stay open.
            pauseDuration: "{{ inputs.approval_timeout }}"
            onResume:
              - id: approved
                type: BOOL
                displayName: Backfill the missing required field?
                defaults: false
              - id: note
                type: STRING
                displayName: Decision note
                defaults: ""

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: repair_branch
                type: io.kestra.plugin.core.flow.If
                description: Only open the write path when the detection found something a
                  default can actually repair.
                condition: "{{ (outputs.detect.rows[0].targeted ?? 0) > 0 }}"
                then:
                  - id: remediate
                    type: io.kestra.plugin.core.flow.Loop
                    description: >-
                      One batch of document ids per iteration. concurrencyLimit
                      1 keeps the batches sequential, because parallel writes to
                      the same collection under a validator race for no gain.
                    concurrencyLimit: 1
                    values: "{{ render(vars.repair_batches) }}"
                    tasks:
                      - id: backfill
                        type: io.kestra.plugin.mongodb.Update
                        description: >-
                          One UPDATE_MANY per batch. The filter is the ids the
                          detection classified as repairable, re-asserting that
                          the field is still absent and that the document still
                          satisfies the rest of the schema. A document that
                          changed since the detection ran is skipped rather than
                          written over.
                        connection:
                          uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                            secret('MONGODB_URI') }}"
                        database: "{{ inputs.database }}"
                        collection: "{{ inputs.collection }}"
                        operation: UPDATE_MANY
                        filter: >-
                          {{ {'ids': item.value, 'f': inputs.required_field,
                          's': fromJson(inputs.json_schema)} | jq('.f as $f | .s
                          as $s | ($s | .required = ((($s.required) // []) |
                          map(select(. != $f)))) as $sm | {"$and": [{"_id":
                          {"$in": (.ids | fromjson)}}, {($f): {"$exists":
                          false}}, {"$jsonSchema": $sm}]}') | first | toJson }}
                        document: "{{ render(vars.set_document) }}"

                  - id: repaired_check
                    type: io.kestra.plugin.mongodb.Aggregate
                    description: >-
                      Read-only. Measures the exact documents the repair
                      targeted, by id. How many now match the full schema, how
                      many still have no value in the field, plus the collection
                      wide compliant count.
                    connection:
                      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                        secret('MONGODB_URI') }}"
                    database: "{{ inputs.database }}"
                    collection: "{{ inputs.collection }}"
                    pipeline: "{{ render(vars.verify_pipeline) }}"
                    store: FETCH

                  - id: log_repaired
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Of the {{ outputs.detect.rows[0].targeted ?? 0 }} targeted
                      document(s), {{
                      outputs.repaired_check.rows[0].targeted_valid ?? 0 }} now
                      match the full schema and {{
                      outputs.repaired_check.rows[0].targeted_missing ?? 0 }}
                      still have no {{ inputs.required_field }}. The collection
                      now holds {{ outputs.repaired_check.rows[0].compliant ?? 0
                      }} compliant document(s).

                  - id: reverify
                    type: io.kestra.plugin.mongodb.Find
                    description: >-
                      Read-only. The detection predicate again, collection wide.
                      Any document that still fails the schema only because this
                      field is missing comes back here, whether the write missed
                      it or a writer created it while the run was paused.
                    connection:
                      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                        secret('MONGODB_URI') }}"
                    database: "{{ inputs.database }}"
                    collection: "{{ inputs.collection }}"
                    filter: "{{ render(vars.repairable_filter) }}"
                    limit: "{{ inputs.sample_limit }}"
                    store: false

                  - id: log_reverify
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Re-check after the backfill found {{ outputs.reverify.size
                      ?? 0 }} document(s) still missing {{ inputs.required_field
                      }} in {{ inputs.database }}.{{ inputs.collection }}.
                      Sample: {{ outputs.reverify.rows[0] ?? 'none' }}.

                  - id: verify_gate
                    type: io.kestra.plugin.core.flow.If
                    description: Fail the run if the collection still holds a repairable violator
                      after the backfill.
                    condition: "{{ (outputs.reverify.size ?? 0) > 0 }}"
                    then:
                      - id: remediation_failed
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: >-
                          The backfill ran but the re-check still finds {{
                          outputs.reverify.size ?? 0 }} document(s) in {{
                          inputs.database }}.{{ inputs.collection }} that fail
                          the validator with no {{ inputs.required_field }}. The
                          write may not have landed. The repair_limit of {{
                          inputs.repair_limit }} may not cover the whole drift.
                          Something may still be writing documents without that
                          field. The collection does not match its own
                          validator.
                    else:
                      - id: remediated
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Backfilled {{ inputs.required_field }} on {{
                          outputs.repaired_check.rows[0].targeted_valid ?? 0 }}
                          document(s) in {{ inputs.database }}.{{
                          inputs.collection }}. The re-check finds {{
                          outputs.reverify.size ?? 0 }} document(s) still
                          missing it, so the missing field drift is cleared. The
                          {{ outputs.detect.rows[0].unrepairable ?? 0 }} refused
                          document(s) were not touched.
                else:
                  - id: nothing_to_repair
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      {{ outputs.detect.rows[0].violators ?? 0 }} document(s)
                      fail the validator but none of them can be fixed by
                      backfilling {{ inputs.required_field }}. Nothing was
                      written. These need a value only a human can decide. See
                      the refused list in the report.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Approval not granted{{ (outputs.hold.onResume.note ?? '') ==
                  '' ? '' : (', note: ' ~ outputs.hold.onResume.note) }}.
                  Nothing was written. The documents that fail the validator
                  stay as they are.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: >-
          No validation drift. Every document in {{ inputs.database }}.{{
          inputs.collection }} matches the schema its validator enforces.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': ('MongoDB validation drift repair guard failed on ' ~ inputs.database ~ '.' ~ inputs.collection ~ ' (execution ' ~ execution.id ~ '). Check the logs.')} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Shipped disabled. Its inputs force safe mode, so even an enabled schedule
      only ever reports and never writes on its own.
    cron: "0 5 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

A $jsonSchema validator added to a collection that already holds data applies to new writes only. MongoDB never goes back over what is already there. So every document written before the validator keeps breaking the rule nobody checks: a required field is missing or a field holds the wrong BSON type. Readers that trust the schema break on those documents. The day somebody runs a migration that touches one, the write fails with error 121 even though the migration had nothing to do with the broken field.

This blueprint reads the collection with $nor on the schema, which is exactly the set of documents the validator would reject today. Then it splits that set in two. A document whose only problem is the one missing required field can be repaired from a default. Everything else is reported with its BSON value type and its missing required keys, then left alone. Behind a safe mode gate and an approval it backfills the repairable documents with $set, in batches of document ids, reads the collection again and fails the run if any repairable violator is still there.

This blueprint was created by zkasuran.

Safety

Read-only by default. The first run with the defaults only counts and reports. A write needs auto_remediate true, dry_run false, then a human answering the Pause. Three conditions, all of them, before a document is touched.
What it refuses to repair is the point. A backfill can supply a default for a field that is absent. It cannot invent the value that field should have held. It must never coerce a field that holds the wrong type, because a guessed cast writes plausible wrong data into production and that is worse than the broken document it replaces. Those documents are counted, named with their BSON type, then left for a human.
The input check runs before the first read. A required_field the schema does not require is refused. A default_value whose JSON type contradicts the bsonType the schema declares for that field is refused. A repair that writes a number into a string field only trades one broken rule for another.
The write is scoped to the document ids the detection classified, in batches of batch_size, one batch at a time. A document that changed since the detection ran no longer matches the filter, so it is skipped rather than written over. Drift created while the run was paused is never swept in silently. It shows up in the re-read instead.
The re-read is the proof, not the modified count. If the collection still holds a document that fails the validator only because the field is missing, the run fails.

How it works

check_inputs (io.kestra.plugin.core.flow.If) refuses a required_field that is not in the schema required list and a default_value whose JSON type contradicts the declared bsonType. It fails the run before the first read.
detect (io.kestra.plugin.mongodb.Aggregate) matches {"$nor": [{"$jsonSchema": schema}]}, then counts in one pass: every violator, the ones repairable by backfilling the field, the capped list of their ids, plus a sample of the refused documents carrying the $type of the field and the required keys they are missing. Repairability is decided by testing the document against the same schema with the target field dropped from required, so a document that is broken in a second way is never claimed as repairable.
report (io.kestra.plugin.core.log.Log) logs the counts. Task outputs are not in the execution JSON, so the numbers the gates act on are written to the log.
route (io.kestra.plugin.core.flow.If): zero violators logs clean and stops. One or more sends alert to Slack and prints planned, which states what a repair run would do on every run including a dry one.
remediate_gate (io.kestra.plugin.core.flow.If) opens only with auto_remediate true and dry_run false. hold (io.kestra.plugin.core.flow.Pause) waits for an answer.
On approval remediate (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) walks the id batches and backfill (io.kestra.plugin.mongodb.Update, UPDATE_MANY) applies the $set. With nothing repairable the run logs nothing_to_repair and writes nothing.
repaired_check (io.kestra.plugin.mongodb.Aggregate) measures the targeted ids, reverify (io.kestra.plugin.mongodb.Find) runs the detection predicate across the collection again and verify_gate fails the run with io.kestra.plugin.core.execution.Fail if anything repairable remains. Without approval declined records the decision.
errors posts to Slack when any task fails, for example when MongoDB is unreachable or the validator rejects the repair itself.

What you get

The exact count of documents your own validator would reject right now, which no MongoDB command reports for you.
That count split into what a default can fix and what needs a person, with the refused documents named by id, by the BSON type of the field and by the required keys they lack.
A gated, approved backfill of only the repairable documents, in bounded batches.
A re-read that fails the run when the collection still does not match its own validator, so a partial write or a writer still producing drift is caught rather than hidden.

Who it's for

Teams who added a validator to a live collection and never cleaned up what predates it.
Anyone whose migration failed with error 121 on documents that were already broken.
Platform engineers who want schema drift reported on a schedule with an audit trail per run.

Why orchestrate this with Kestra

MongoDB enforces a validator at write time and offers nothing to audit what is already stored. There is no command that lists the documents a validator would reject, so the usual answer is a one-off script somebody runs by hand once. This flow makes it a scheduled, reviewable job. The detection is one server-side pipeline, the repair sits behind auto_remediate, dry_run and a Pause with the decision recorded on the execution, the batches are serialised by concurrencyLimit: 1 and the re-read turns the write into a checked outcome. The Slack alert and the error path are the same ones the rest of your flows use.

Tested end to end

On Kestra 2.0.5 OSS against a MongoDB 7.0 container, on the final file. appdata.orders held 13 documents written before the validator was added with collMod: 6 missing the required status field, 4 fully compliant, 2 holding status as an int and a bool, 1 missing both required fields. The validator requires order_id and status as strings. Every before and after number was read back with mongosh, not only from the Kestra log.

Run	Inputs	Result
Clean	a collection with 3 compliant documents	0 violators. `clean` logged, no Slack alert, SUCCESS
Detect	defaults	9 violators, 6 repairable, 3 refused with types `int`, `bool`, missing keys. Slack alert, SUCCESS
Dry run	defaults	Reports the repair it would make. `mongosh` still counts 7 documents with no `status`
Armed but dry	`auto_remediate: true`, `dry_run: true`	Gate stayed closed, no `Pause`, nothing written
Declined	`auto_remediate: true`, `dry_run: false`, resumed `approved: false`	Paused, then `declined`. Nothing written
Repair	the same, resumed `approved: true`	6 backfilled. `mongosh`: violators 9 to 3, compliant 4 to 10, the 4 compliant documents byte for byte unchanged
Nothing repairable	run again on the repaired collection	3 violators, 0 repairable, `nothing_to_repair`, no write
Re-verify fails	validator set to `validationAction: warn`, a writer unset `status` on a compliant document during the `Pause`	The 6 targeted documents were repaired, the collection wide re-read found 1 left, `verify_gate` FAILED the run and `errors` alerted
Bad default	`default_value: 0` against a `string` field	Refused before any read
Wrong field	`required_field: total`, which the schema does not require	Refused before any read
Unreachable	`connection_uri` pointed at a dead host	`detect` failed, `errors` posted to Slack, execution FAILED

Does the detector actually fail

Four ways, all measured. The clean collection returned 0 while the drifted one returned 9 with nothing else changed. The refused class is not a rounding error: 3 documents the flow could have quietly coerced were counted, named by BSON type and left alone, which mongosh confirms after the repair run. The repair itself is checked twice, by id and collection wide. And the gate was forced to fire by unsetting a required field on a compliant document while the run was paused, which the id scoped write correctly ignored, so only the re-read caught it and the run failed.

The real world bite was reproduced too. With a violator present, updateOne setting an unrelated field on it was rejected with error 121 by the validator, because the result still failed the schema. After the repair run the identical update succeeded.

Prerequisites

A reachable MongoDB instance and a user that can read the collection and run an update on it.
A collection with a $jsonSchema validator. Read the schema with db.getCollectionInfos({name: 'orders'})[0].options.validator and pass the $jsonSchema object as the json_schema input.
A value worth writing into the missing field. If no safe default exists, run the flow read-only and use the report.
Local testing: docker run -d --name mdb --network mynet mongo:7.0, insert documents, add the validator with collMod afterwards, then run the flow with connection_uri: mongodb://mdb:27017.

Secrets

MONGODB_URI: the MongoDB connection string. Read when connection_uri is left empty, which is how the flow should run in production. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_MONGODB_URI.
SLACK_WEBHOOK_URL: used by alert and the errors block. Provide it as SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`connection_uri`	empty	Plain connection string for a local test. Empty reads `MONGODB_URI`
`database`, `collection`	`appdata`, `orders`	The collection to audit
`json_schema`	the worked example	The `$jsonSchema` the collection is validated against
`required_field`	`status`	The one missing required field this run backfills
`default_value`	`"unknown"`	JSON literal written into that field. Quote a string value
`repair_limit`	`1000`	Ceiling on documents repaired per run
`batch_size`	`250`	Document ids per write batch
`sample_limit`	`25`	Refused documents named per report, also the re-check read cap
`approval_timeout`	`PT2H`	How long the pause waits for an answer
`auto_remediate`	`false`	Allow the backfill
`dry_run`	`true`	Must be `false` for a write to apply

Quick start

Add the MONGODB_URI and SLACK_WEBHOOK_URL secrets.
Paste your collection $jsonSchema into json_schema, set database and collection, then run the flow. It reports the drift and stops.
Read the refused list. Decide whether a default is right for the field you are backfilling.
Set required_field with its default_value, then auto_remediate: true and dry_run: false, run it again and answer the pause. The backfill runs and the re-read proves the repairable drift is gone.
Enable the daily schedule, which always runs in safe mode.

How to extend

Audit several collections by wrapping the tasks in a io.kestra.plugin.core.flow.Loop over a list of collection and schema pairs, with concurrencyLimit: 1 on the part that writes.
Read the schema from the collection itself with a runCommand style task instead of passing it as an input, once your plugin build ships one, so the audit can never drift from the live validator.
Repair more than one missing field in a run by extending the $set document. Apply them in one write, because under validationLevel: strict an intermediate document that is still invalid is rejected.
Route the refused documents to a ticket instead of a log line, so the ones needing a human decision get an owner.
Once the collection is clean, move the validator to validationLevel: strict with validationAction: error so the drift cannot come back.

MongoDB Document Validation Drift Repair Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Does the detector actually fail

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MongoDB Document Validation Drift Repair Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Does the detector actually fail

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

MinIO Cluster Disk Healing Sentinel

Enforce default SSE-S3 encryption on a MinIO bucket

Meilisearch Indexing Queue Staleness Sentinel

Incremental, idempotent lake ingestion from MinIO with DuckDB

Catch MinIO Replication Drift Before Objects Go Missing on the Target

Enforce a MinIO bucket lifecycle expiry and noncurrent version policy

More Data Blueprints

Huawei GeminiDB TTL Hygiene Sentinel

Idempotent Incremental CSV Loader from S3 to PostgreSQL

Sync Hubspot CRM data to BigQuery on a schedule

InfluxDB Bucket Retention Sentinel

Automate Apache Iceberg Table Maintenance and Snapshot Expiry via REST Catalog

Huawei DLI Data Quality Assertion Gate

New to Kestra?