Backfill OpenSearch Documents Hidden by Bulk Rejections

id: opensearch-missing-document-bulk-backfill
namespace: company.team
description: |
  Find documents that a successful-looking OpenSearch Bulk response did not index,
  restore only those documents from PostgreSQL and prove the window has conserved ids.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: opensearch_url
    type: STRING
    defaults: "{{ secret('OPENSEARCH_HOST') }}"
    description: OpenSearch URL reachable by the Kestra worker.
  - id: opensearch_username
    type: STRING
    defaults: "{{ secret('OPENSEARCH_USERNAME') }}"
    description: OpenSearch username with read and index privileges on the target index.
  - id: opensearch_password
    type: STRING
    defaults: "{{ secret('OPENSEARCH_PASSWORD') }}"
    description: OpenSearch password.
  - id: jdbc_url
    type: STRING
    defaults: "{{ secret('POSTGRES_URL') }}"
    description: PostgreSQL JDBC URL for the source and the dead-letter table.
  - id: postgres_username
    type: STRING
    defaults: "{{ secret('POSTGRES_USERNAME') }}"
    description: PostgreSQL username.
  - id: postgres_password
    type: STRING
    defaults: "{{ secret('POSTGRES_PASSWORD') }}"
    description: PostgreSQL password.
  - id: index_name
    type: STRING
    defaults: orders
    description: One OpenSearch index whose documents contain the id and bucket fields.
  - id: source_table
    type: STRING
    defaults: public.search_documents
    description: PostgreSQL table or view containing the authoritative documents.
  - id: window_start
    type: STRING
    defaults: "2026-01-01T00:00:00Z"
    description: Inclusive lower bound for the bucket window.
  - id: window_end
    type: STRING
    defaults: "2026-01-01T01:00:00Z"
    description: Exclusive upper bound for the bucket window.
  - id: bucket_values
    type: ARRAY
    itemType: STRING
    defaults:
      - "2026-01-01T00:00:00Z"
    description: Exact bucket values to count with Search after the replay. Keep
      this list bounded.
  - id: max_rows
    type: INT
    defaults: 1000
    description: Maximum source rows in one run. The flow refuses a truncated snapshot.
  - id: bulk_chunk
    type: INT
    defaults: 100
    description: Number of Bulk operations per OpenSearch request.
  - id: approval_timeout
    type: STRING
    defaults: PT10M
    description: How long the approval remains open.
  - id: dry_run
    type: BOOL
    defaults: true
    description: Report the gap without writing when true.
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: Arm the approval and Bulk write only when true and dry_run is false.

triggers:
  - id: hourly_window
    type: io.kestra.plugin.core.trigger.Schedule
    description: Hourly backstop in safe report-only mode. Set the window inputs for
      a real run.
    cron: "0 * * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: source_stats
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Check the source window has unique non-null ids and fits the
      bounded snapshot.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.postgres_username }}"
    password: "{{ inputs.postgres_password }}"
    fetchType: FETCH_ONE
    parameters:
      start: "{{ inputs.window_start }}"
      end: "{{ inputs.window_end }}"
    sql: |
      SELECT count(*) AS source_count,
             count(*) FILTER (WHERE id IS NULL) AS null_ids,
             count(*) - count(DISTINCT id) AS duplicate_ids
      FROM {{ inputs.source_table }}
      WHERE CAST(bucket AS text) >= CAST(:start AS text)
        AND CAST(bucket AS text) < CAST(:end AS text)

  - id: source_guard
    type: io.kestra.plugin.core.flow.If
    description: Refuse null or duplicate ids and refuse a window larger than the
      declared ceiling.
    condition: "{{ outputs.source_stats.row.null_ids > 0 or
      outputs.source_stats.row.duplicate_ids > 0 or
      outputs.source_stats.row.source_count > inputs.max_rows }}"
    then:
      - id: fail_source_guard
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Source window is not a safe bounded id set: {{
          outputs.source_stats.row.source_count }} rows, {{
          outputs.source_stats.row.null_ids }} null ids and {{
          outputs.source_stats.row.duplicate_ids }} duplicate ids (max_rows={{
          inputs.max_rows }})."

  - id: source_rows
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Store the authoritative rows for this bounded window before
      comparing the index.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.postgres_username }}"
    password: "{{ inputs.postgres_password }}"
    fetchType: STORE
    parameters:
      start: "{{ inputs.window_start }}"
      end: "{{ inputs.window_end }}"
      limit: "{{ inputs.max_rows + 1 }}"
    sql: |
      SELECT *
      FROM {{ inputs.source_table }}
      WHERE CAST(bucket AS text) >= CAST(:start AS text)
        AND CAST(bucket AS text) < CAST(:end AS text)
      ORDER BY id
      LIMIT CAST(:limit AS integer)

  - id: index_scroll
    type: io.kestra.plugin.opensearch.Scroll
    description: Scroll the complete id set for the same window. The source keeps
      the id field in _source.
    connection:
      hosts:
        - "{{ inputs.opensearch_url }}"
      basicAuth:
        username: "{{ inputs.opensearch_username }}"
        password: "{{ inputs.opensearch_password }}"
    indexes:
      - "{{ inputs.index_name }}"
    request:
      size: 500
      _source:
        - "id"
        - "bucket"
      query:
        match_all: {}

  - id: source_json
    type: io.kestra.plugin.serdes.json.IonToJson
    description: Convert the stored PostgreSQL rows into JSONL for the bounded diff.
    from: "{{ outputs.source_rows.uri }}"

  - id: index_json
    type: io.kestra.plugin.serdes.json.IonToJson
    description: Convert the Scroll result into JSONL so ids can be compared without
      a custom OpenSearch client.
    from: "{{ outputs.index_scroll.uri }}"

  - id: compute_gap
    type: io.kestra.plugin.scripts.python.Script
    description: Compute missing ids, conserved bucket counts and the exact source
      rows eligible for replay.
    containerImage: python:3.13-slim
    inputFiles:
      source.jsonl: "{{ outputs.source_json.uri }}"
      index.jsonl: "{{ outputs.index_json.uri }}"
    outputFiles:
      - gap.json
    env:
      ID_FIELD: "id"
      BUCKET_FIELD: "bucket"
      WINDOW_START: "{{ inputs.window_start }}"
      WINDOW_END: "{{ inputs.window_end }}"
    script: |
      import json
      import os
      import sys
      from collections import Counter

      def records(path):
          text = open(path, encoding="utf-8").read().strip()
          if not text:
              return []
          try:
              value = json.loads(text)
              return value if isinstance(value, list) else [value]
          except json.JSONDecodeError:
              return [json.loads(line) for line in text.splitlines() if line.strip()]

      id_field = os.environ["ID_FIELD"]
      bucket_field = os.environ["BUCKET_FIELD"]
      source = records("source.jsonl")
      all_indexed = records("index.jsonl")
      indexed = [row for row in all_indexed if os.environ["WINDOW_START"] <= str(row.get(bucket_field)) < os.environ["WINDOW_END"]]
      source_ids = [str(row.get(id_field)) for row in source]
      index_ids = [str(row.get(id_field)) for row in indexed]
      if any(value == "None" for value in source_ids + index_ids):
          sys.exit("The source or Scroll _source omitted the configured id field.")
      if len(source_ids) != len(set(source_ids)):
          sys.exit("The source snapshot contains duplicate ids.")
      if len(index_ids) != len(set(index_ids)):
          sys.exit("The OpenSearch window contains duplicate ids.")
      source_map = {str(row[id_field]): row for row in source}
      source_set = set(source_map)
      index_set = set(index_ids)
      missing = sorted(source_set - index_set)
      unexpected = sorted(index_set - source_set)
      source_buckets = Counter(str(row.get(bucket_field)) for row in source)
      index_buckets = Counter(str(row.get(bucket_field)) for row in indexed)
      gap = {
          "source_count": len(source),
          "index_count": len(indexed),
          "source_snapshot": source,
          "missing_ids": missing,
          "unexpected_ids": unexpected,
          "source_buckets": dict(source_buckets),
          "index_buckets": dict(index_buckets),
          "replay_rows": [source_map[value] for value in missing],
      }
      json.dump(gap, open("gap.json", "w", encoding="utf-8"), sort_keys=True)
      print("Source ids: %d. Indexed ids: %d. Missing: %d. Unexpected in window: %d." % (len(source), len(indexed), len(missing), len(unexpected)))
      print("Bucket counts before repair: %s" % json.dumps(dict(source_buckets)))
      print("::" + json.dumps({"outputs": {"source_count": len(source), "index_count": len(indexed), "missing_count": len(missing), "unexpected_count": len(unexpected)}}) + "::")

  - id: report_gap
    type: io.kestra.plugin.core.log.Log
    description: Record the measured before state and the exact candidate ids.
    message: "OpenSearch {{ inputs.index_name }} window {{ inputs.window_start }} to
      {{ inputs.window_end }}: source={{ outputs.compute_gap.vars.source_count
      }}, indexed={{ outputs.compute_gap.vars.index_count }}, missing={{
      outputs.compute_gap.vars.missing_count }}, unexpected={{
      outputs.compute_gap.vars.unexpected_count }}. Dry run={{ inputs.dry_run }}
      auto_remediate={{ inputs.auto_remediate }}."

  - id: alert_gap
    type: io.kestra.plugin.core.flow.If
    description: Alert when the source/index count gap is found, before deciding
      whether to write.
    condition: "{{ outputs.compute_gap.vars.missing_count > 0 }}"
    then:
      - id: slack_gap
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text":"OpenSearch missing-document gap in {{ inputs.index_name }}: {{ outputs.compute_gap.vars.missing_count }} source ids absent from the {{ inputs.window_start }} to {{ inputs.window_end }} window. Bulk may have returned HTTP 200 with item rejections. Remediation={{ inputs.auto_remediate and not inputs.dry_run }}."}

  - id: build_bulk
    type: io.kestra.plugin.scripts.python.Script
    description: Render only the missing source rows as idempotent index operations
      for the OpenSearch Bulk task.
    containerImage: python:3.13-slim
    inputFiles:
      gap.json: "{{ outputs.compute_gap.outputFiles['gap.json'] }}"
    outputFiles:
      - missing.ndjson
    env:
      INDEX_NAME: "{{ inputs.index_name }}"
      ID_FIELD: "id"
    script: |
      import json
      gap = json.load(open("gap.json", encoding="utf-8"))
      with open("missing.ndjson", "w", encoding="utf-8") as out:
          for row in gap["replay_rows"]:
              out.write(json.dumps({"index": {"_index": __import__("os").environ["INDEX_NAME"], "_id": str(row[__import__("os").environ["ID_FIELD"]])}}) + "\n")
              out.write(json.dumps(row, separators=(",", ":")) + "\n")
      print("Prepared %d idempotent index operations." % len(gap["replay_rows"]))

  - id: remediation_gate
    type: io.kestra.plugin.core.flow.If
    description: Only a real armed run with a non-empty gap reaches approval and Bulk.
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      outputs.compute_gap.vars.missing_count > 0 }}"
    then:
      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        description: Approve replaying exactly the missing ids listed in the execution
          log. No orphan or stale document is deleted.
        pauseDuration: "{{ inputs.approval_timeout }}"
        behavior: FAIL
        onResume:
          - id: approved
            type: BOOL
            required: false
            description: Set true to re-read both systems and replay the planned missing
              ids.

      - id: approval_gate
        type: io.kestra.plugin.core.flow.If
        description: An unanswered or declined approval never writes.
        condition: "{{ outputs.approval.onResume.approved ?? false }}"
        then:
          - id: source_recheck
            type: io.kestra.plugin.jdbc.postgresql.Query
            description: Re-read the source immediately before Bulk so the approval is bound
              to current rows.
            url: "{{ inputs.jdbc_url }}"
            username: "{{ inputs.postgres_username }}"
            password: "{{ inputs.postgres_password }}"
            fetchType: STORE
            parameters:
              start: "{{ inputs.window_start }}"
              end: "{{ inputs.window_end }}"
              limit: "{{ inputs.max_rows + 1 }}"
            sql: |
              SELECT * FROM {{ inputs.source_table }}
              WHERE CAST(bucket AS text) >= CAST(:start AS text)
                AND CAST(bucket AS text) < CAST(:end AS text)
              ORDER BY id
              LIMIT CAST(:limit AS integer)
          - id: index_recheck_scroll
            type: io.kestra.plugin.opensearch.Scroll
            description: Re-scroll the index after approval. A changed candidate set
              invalidates the plan.
            connection:
              hosts:
                - "{{ inputs.opensearch_url }}"
              basicAuth:
                username: "{{ inputs.opensearch_username }}"
                password: "{{ inputs.opensearch_password }}"
            indexes:
              - "{{ inputs.index_name }}"
            request:
              size: 500
              _source:
                - "id"
                - "bucket"
              query:
                match_all: {}
          - id: source_recheck_json
            type: io.kestra.plugin.serdes.json.IonToJson
            from: "{{ outputs.source_recheck.uri }}"
          - id: index_recheck_json
            type: io.kestra.plugin.serdes.json.IonToJson
            from: "{{ outputs.index_recheck_scroll.uri }}"
          - id: stale_plan_check
            type: io.kestra.plugin.scripts.python.Script
            description: Refuse the write if either source rows or current index ids differ
              from the approved snapshot.
            containerImage: python:3.13-slim
            inputFiles:
              approved.json: "{{ outputs.compute_gap.outputFiles['gap.json'] }}"
              source.jsonl: "{{ outputs.source_recheck_json.uri }}"
              index.jsonl: "{{ outputs.index_recheck_json.uri }}"
            env:
              ID_FIELD: "id"
              WINDOW_START: "{{ inputs.window_start }}"
              WINDOW_END: "{{ inputs.window_end }}"
            script: |
              import json, os, sys
              def records(path):
                  text = open(path, encoding="utf-8").read().strip()
                  if not text: return []
                  try:
                      value = json.loads(text)
                      return value if isinstance(value, list) else [value]
                  except json.JSONDecodeError:
                      return [json.loads(line) for line in text.splitlines() if line.strip()]
              field = os.environ["ID_FIELD"]
              approved = json.load(open("approved.json", encoding="utf-8"))
              source = {str(row[field]) for row in records("source.jsonl")}
              indexed = {str(row[field]) for row in records("index.jsonl") if os.environ["WINDOW_START"] <= str(row.get("bucket")) < os.environ["WINDOW_END"]}
              approved_source = {json.dumps(row, sort_keys=True, default=str) for row in approved["source_snapshot"]}
              current_source = {json.dumps(row, sort_keys=True, default=str) for row in records("source.jsonl")}
              current_missing = sorted(source - indexed)
              if current_source != approved_source or current_missing != approved["missing_ids"]:
                  sys.exit("STALE PLAN: source or indexed id set changed during approval; no Bulk write sent.")
              print("Approved id set is unchanged: %d missing ids." % len(current_missing))
          - id: reindex_missing
            type: io.kestra.plugin.opensearch.Bulk
            description: Replay only the approved source rows. Stable ids make a retry
              idempotent.
            connection:
              hosts:
                - "{{ inputs.opensearch_url }}"
              basicAuth:
                username: "{{ inputs.opensearch_username }}"
                password: "{{ inputs.opensearch_password }}"
            from: "{{ outputs.build_bulk.outputFiles['missing.ndjson'] }}"
            chunk: "{{ inputs.bulk_chunk }}"
          - id: refresh_index
            type: io.kestra.plugin.opensearch.Request
            description: Make the Bulk writes visible to the verification Scroll before
              counting residual ids.
            connection:
              hosts:
                - "{{ inputs.opensearch_url }}"
              basicAuth:
                username: "{{ inputs.opensearch_username }}"
                password: "{{ inputs.opensearch_password }}"
            method: POST
            endpoint: "{{ inputs.index_name }}/_refresh"
        else:
          - id: declined
            type: io.kestra.plugin.core.log.Log
            message: "Approval declined or expired. No Bulk request was sent for {{
              outputs.compute_gap.vars.missing_count }} ids."
    else:
      - id: report_only
        type: io.kestra.plugin.core.log.Log
        description: Safe mode and healthy runs never write to OpenSearch.
        message: "No remediation requested. The measured gap is {{
          outputs.compute_gap.vars.missing_count }} ids."

  - id: verification_route
    type: io.kestra.plugin.core.flow.If
    description: Verify only after an approved Bulk write. Report-only runs stop
      after the measured before state.
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      outputs.compute_gap.vars.missing_count > 0 and
      (outputs.approval.onResume.approved ?? false) }}"
    then:
      - id: verify_source
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "{{ inputs.jdbc_url }}"
        username: "{{ inputs.postgres_username }}"
        password: "{{ inputs.postgres_password }}"
        fetchType: STORE
        parameters:
          start: "{{ inputs.window_start }}"
          end: "{{ inputs.window_end }}"
          limit: "{{ inputs.max_rows + 1 }}"
        sql: |
          SELECT * FROM {{ inputs.source_table }}
          WHERE CAST(bucket AS text) >= CAST(:start AS text)
            AND CAST(bucket AS text) < CAST(:end AS text)
          ORDER BY id
          LIMIT CAST(:limit AS integer)
      - id: verify_scroll
        type: io.kestra.plugin.opensearch.Scroll
        connection:
          hosts:
            - "{{ inputs.opensearch_url }}"
          basicAuth:
            username: "{{ inputs.opensearch_username }}"
            password: "{{ inputs.opensearch_password }}"
        indexes:
          - "{{ inputs.index_name }}"
        request:
          size: 500
          _source:
            - "id"
            - "bucket"
          query:
            match_all: {}
      - id: verify_source_json
        type: io.kestra.plugin.serdes.json.IonToJson
        from: "{{ outputs.verify_source.uri }}"
      - id: verify_index_json
        type: io.kestra.plugin.serdes.json.IonToJson
        from: "{{ outputs.verify_scroll.uri }}"
      - id: verify_gap
        type: io.kestra.plugin.scripts.python.Script
        description: Compute residual ids after remediation and fail only after the
          residual list is persisted.
        containerImage: python:3.13-slim
        inputFiles:
          source.jsonl: "{{ outputs.verify_source_json.uri }}"
          index.jsonl: "{{ outputs.verify_index_json.uri }}"
        outputFiles:
          - residual.json
        env:
          ID_FIELD: "id"
          BUCKET_FIELD: "bucket"
          WINDOW_START: "{{ inputs.window_start }}"
          WINDOW_END: "{{ inputs.window_end }}"
        script: |
          import json, os
          def records(path):
              text = open(path, encoding="utf-8").read().strip()
              if not text: return []
              try:
                  value = json.loads(text)
                  return value if isinstance(value, list) else [value]
              except json.JSONDecodeError:
                  return [json.loads(line) for line in text.splitlines() if line.strip()]
          field = os.environ["ID_FIELD"]
          source = {str(row[field]) for row in records("source.jsonl")}
          indexed = {str(row[field]) for row in records("index.jsonl") if os.environ["WINDOW_START"] <= str(row.get("bucket")) < os.environ["WINDOW_END"]}
          residual = sorted(source - indexed)
          json.dump(residual, open("residual.json", "w", encoding="utf-8"))
          print("After Bulk: source ids=%d, indexed ids=%d, residual ids=%d." % (len(source), len(indexed), len(residual)))
          print("::" + json.dumps({"outputs": {"residual_count": len(residual), "verified_index_count": len(indexed), "verified_source_count": len(source)}}) + "::")
      - id: verify_search_total
        type: io.kestra.plugin.opensearch.Search
        description: Ask the OpenSearch Search task for the complete index count as an
          independent system-owned read.
        connection:
          hosts:
            - "{{ inputs.opensearch_url }}"
          basicAuth:
            username: "{{ inputs.opensearch_username }}"
            password: "{{ inputs.opensearch_password }}"
        indexes:
          - "{{ inputs.index_name }}"
        fetchType: FETCH
        request:
          size: 0
          query:
            match_all: {}
      - id: verify_bucket_counts
        type: io.kestra.plugin.core.flow.Loop
        description: Search each declared id bucket with size zero so OpenSearch reports
          conserved counts independently of the Scroll diff.
        values: "{{ inputs.bucket_values | toJson }}"
        concurrencyLimit: 1
        tasks:
          - id: search_bucket
            type: io.kestra.plugin.opensearch.Request
            connection:
              hosts:
                - "{{ inputs.opensearch_url }}"
              basicAuth:
                username: "{{ inputs.opensearch_username }}"
                password: "{{ inputs.opensearch_password }}"
            method: POST
            endpoint: "{{ inputs.index_name }}/_search"
            body: |
              {"size":0,"query":{"term":{"bucket":"{{ item.value }}"}}}
          - id: log_bucket_count
            type: io.kestra.plugin.core.log.Log
            message: "Search count for bucket {{ item.value }} after repair: {{
              outputs.search_bucket.response.hits.total.value }}"
      - id: bucket_loop_report
        type: io.kestra.plugin.core.log.Log
        message: "Bucket Search iterations completed: {{ outputs.verify_bucket_counts
          }}"
      - id: residual_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.verify_gap.vars.residual_count > 0 }}"
        then:
          - id: ensure_dead_letter
            type: io.kestra.plugin.jdbc.postgresql.Queries
            url: "{{ inputs.jdbc_url }}"
            username: "{{ inputs.postgres_username }}"
            password: "{{ inputs.postgres_password }}"
            fetchType: NONE
            parameters:
              execution: "{{ execution.id }}"
              index_name: "{{ inputs.index_name }}"
              start: "{{ inputs.window_start }}"
              end: "{{ inputs.window_end }}"
              residual: "{{ read(outputs.verify_gap.outputFiles['residual.json']) }}"
            sql: |
              CREATE TABLE IF NOT EXISTS opensearch_missing_document_dead_letter (
                execution_id TEXT NOT NULL,
                index_name TEXT NOT NULL,
                document_id TEXT NOT NULL,
                window_start TEXT NOT NULL,
                window_end TEXT NOT NULL,
                recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                PRIMARY KEY (execution_id, index_name, document_id)
              );
              INSERT INTO opensearch_missing_document_dead_letter (execution_id, index_name, document_id, window_start, window_end)
              SELECT CAST(:execution AS text), CAST(:index_name AS text), value, CAST(:start AS text), CAST(:end AS text)
              FROM jsonb_array_elements_text(CAST(:residual AS jsonb));
          - id: fail_residual
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "OpenSearch gap remains after Bulk: {{
              outputs.verify_gap.vars.residual_count }} residual ids were
              written to opensearch_missing_document_dead_letter."
        else:
          - id: verify_success
            type: io.kestra.plugin.core.log.Log
            message: "Backfill closed the gap: source={{
              outputs.verify_gap.vars.verified_source_count }}, indexed={{
              outputs.verify_gap.vars.verified_index_count }}, residual=0.
              Search bucket counts were recorded above."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert the platform team when detection, stale-plan protection, Bulk
      or residual verification fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text":"opensearch-missing-document-bulk-backfill failed in execution {{ execution.id }} for {{ inputs.index_name }} window {{ inputs.window_start }} to {{ inputs.window_end }}. Read the execution logs and the dead-letter table before retrying."}