Sub-Millisecond Event-Driven Ingestion: Announcing the gRPC Push Repository Connector

1. Beyond Polling: The Case for Binary Push Ingestion

Earlier this year, we introduced OpenCrawling's gRPC Internal Transport Layer (oc-grpc-api) to accelerate inter-service communication between the crawler core and decoupled workers. Immediately, enterprise architects asked the natural follow-up question: "Can our external event producers push documents directly into OpenCrawling over gRPC with the same binary efficiency?"

In large-scale production environments, waiting for a 15-minute or hourly polling window is often too slow—and pushing multi-megabyte binary files encoded in Base64 over HTTP/1.1 JSON REST endpoints introduces severe serialization overhead:

• Change Data Capture (CDC) & Event Streams: Debezium connectors, Kafka Connect sinks, and transactional outbox workers emit real-time row mutations and document updates that require immediate vector index synchronization.
• Eliminating Base64 JSON Bloat: Encoding binary PDFs, DOCX contracts, or media assets into JSON strings inflates wire sizes by ~33% and forces heavy CPU allocation during UTF-8 parsing. Protocol Buffers v3 transmit raw binary bytes natively over multiplexed HTTP/2 streams.
• Polyglot Microservice Ecosystems: Enterprise AI platforms rarely run on a single language. Go, Rust, Python, C#, and Node.js services need a strongly-typed, contract-first Protobuf interface to stream documents into OpenCrawling with per-document acknowledgments and reactive backpressure.
• Real-Time Compliance Purges: When a sensitive record is revoked or deleted in an upstream application, compliance mandates require sub-second vector eviction via DELETE tombstones rather than waiting for a nightly reconciliation scan.

2. Four Native gRPC Interaction Models (OisPushService)

Listening on default port 9096, the oc-grpc-push-connector exposes the strongly-typed opencrawling.push.v1.OisPushService contract, supporting four distinct RPC patterns tailored to every workload profile:

The Canonical Protobuf v3 Contract (ois_push_service.proto)

The contract translates the Open Ingestion Standard (OIS) directly into Protocol Buffers v3:

syntax = "proto3";

package opencrawling.push.v1;

service OisPushService {
  rpc PushDocument (PushDocumentRequest) returns (PushDocumentResponse);
  rpc StreamDocuments (stream PushDocumentRequest) returns (StreamSummaryResponse);
  rpc BiDirectionalStream (stream PushDocumentRequest) returns (stream IngestionAck);
  rpc Ping (PingRequest) returns (PingResponse);
}

message PushDocumentRequest {
  string job_id = 1;                           // Target active OpenCrawling Job ID
  OisDocumentPayload document = 2;              // Document content, metadata, and ACLs
  PushOptions options = 3;                      // Pipeline routing and execution options
}

message OisDocumentPayload {
  string id = 1;                               // Unique document identifier (URI or primary key)
  DocumentAction action = 2;                   // UPSERT or DELETE
  string mime_type = 3;                        // MIME type (e.g. text/plain, application/pdf)
  int64 timestamp = 4;                         // Event timestamp (epoch milliseconds)

  oneof content_payload {
    bytes raw_bytes = 5;                       // Inline binary or raw text content
    string claim_check_uri = 6;                // Pre-uploaded claim check URI (file://, ozone://, s3://)
    string text_content = 7;                   // Direct plaintext content
  }

  map<string, string> metadata = 8;            // Dynamic structured metadata attributes
  SecurityConfig security = 9;                 // Document-level Zero-Trust security rules
  string checksum_sha256 = 10;                 // Optional SHA-256 for cryptographic integrity audit
}

message SecurityConfig {
  repeated string allow_sids = 1;              // Security Identifiers allowed to read (users/groups)
  repeated string deny_sids = 2;               // Security Identifiers denied access
  string owner = 3;                            // Document owner / creator SID
  string tenant_id = 4;                        // Multi-tenant isolation boundary
  map<string, string> attributes = 5;          // Custom ABAC attributes
}

3. Java 25 Virtual Threads & Intelligent Claim-Check Offloading

Under the hood, GrpcPushServerManager configures the shaded Netty HTTP/2 server to dispatch every incoming RPC onto a dedicated Java 25 Virtual Thread via Executors.newVirtualThreadPerTaskExecutor(). This allows thousands of concurrent streaming producers to push documents simultaneously without exhausting OS platform threads or blocking Netty event loops.

Even more importantly, the connector protects JVM heap memory and Kafka broker stability through three-tier payload handling in OisDocumentTranslator:

1. Fast-Path In-Memory Streaming (≤ 512 KB): Inline raw_bytes or text_content payloads smaller than inline-binary-threshold-kb (default: 512 KB) flow through a lightweight ByteArrayInputStream directly into the orchestrator without redundant temporary disk writes.
2. Automatic Claim-Check Offloading (> 512 KB): When an inline binary payload (such as a 25 MB PDF report or CAD drawing) exceeds inline-binary-threshold-kb, OisDocumentTranslator automatically streams the bytes into the active ClaimCheckStore (Local storage, Apache Ozone ofs:// / s3g, or AWS S3) and replaces the payload with a compact Claim-Check URI reference.
3. Zero-Copy Pre-Uploaded claim_check_uri Pass-Through: If an upstream producer is handling multi-gigabyte archives or media files that already reside in Apache Ozone or S3, it can simply set claim_check_uri in the Protobuf oneof content_payload. OpenCrawling completely bypasses gRPC binary transfer and routes the URI directly to the decoupled Apache Tika 4 extraction consumers!

4. Instant Tombstone DELETE Short-Circuiting

Keeping enterprise RAG vector stores synchronized with source-of-truth deletions is effortless. When a client pushes a request with action = DELETE:

• OisDocumentTranslator constructs a tombstone RepositoryDocument via RepositoryDocument.createTombstone(id, id).
• JobOrchestrator detects the DELETE lifecycle action and immediately short-circuits Apache Tika text extraction, Mustache narrativization, token chunking, and Ollama/OpenAI embedding generation.
• A lightweight deletion event is published directly to the writer topic, instructing downstream output connectors (pgvector, Milvus, Qdrant, Vespa, OpenSearch, Apache Solr 10, or Luxir) to purge all vector chunks associated with that document ID.

5. Zero-Trust Security, Multi-Tenant ACLs & Mutual TLS (mTLS)

The gRPC Push Connector enforces defense-in-depth at both the transport layer and the document retrieval layer:

1. Transport Security: Constant-Time Bearer Auth & Mutual TLS

• Bearer Token Interceptor: When opencrawling.connector.grpc-push.token is configured, GrpcSecurityInterceptor validates the authorization: Bearer <token> gRPC metadata header using constant-time byte comparison (MessageDigest.isEqual) to prevent timing attacks, while exempting the diagnostic Ping RPC for container health probes.
• TLS & Mutual TLS (mTLS): GrpcPushServerManager integrates Netty's GrpcSslContexts with X.509 certificate chains, PKCS#8 private keys, custom CA trust stores, and configurable client certificate verification (ClientAuth.NONE, OPTIONAL, or REQUIRE).

2. Document-Level Zero-Trust ACL Translation

OisDocumentTranslator.translateSecurity(...) maps the Protobuf SecurityConfig directly into OpenCrawling's unified org.opencrawling.core.security.SecurityConfig:

• allow_sids → PermissionRule(sid, "GROUP", sid, "read")
• deny_sids → PermissionRule(sid, "GROUP", sid, "deny") (preserving explicit deny precedence)
• owner → PermissionRule(owner, "USER", owner, "write")
• tenant_id → PermissionRule(tenantId, "TENANT", tenantId, "read")

When AI agents query the vector index via the OpenCrawling Secure MCP Server, these permissions are strictly enforced at query time—ensuring zero unauthorized data leakage.

6. W3C Distributed Tracing & Lakehouse Audit Ledgers

Pushed documents enjoy the exact same enterprise observability and compliance governance as scheduled repository crawls:

• End-to-End W3C Trace Context Propagation: OpenTelemetryServerInterceptor extracts standard traceparent and tracestate headers from inbound gRPC metadata via GlobalOpenTelemetry.getPropagators(). Upstream microservice spans link seamlessly with OpenCrawling's Kafka producer/consumer spans, Apache Tika extraction spans, and vector indexing spans—visible in the Admin UI Trace Timeline and AIOps Root Cause Analysis engine.
• Automated Lakehouse Parquet Audit Reporting: RuntimePushDocumentDispatcher tracks every unary push execution (exec-push-<8char>) and streaming session (exec-stream-<8char>) in JobLedgerAccumulator. Upon completion, OpenCrawling automatically writes Hive-partitioned Apache Parquet (ZSTD), RFC 4180 CSV Compliance Manifests, and Operational JSON Summaries—immediately queryable via the embedded DuckDB SQL Playground!

7. Developer Experience: oc push CLI, grpcurl & Java 25 Client

We built first-class tooling into the OpenCrawling CLI (oc-cli), the embedded gRPC reflection service, and the Admin Dashboard so you can test and operate gRPC push pipelines in seconds.

1. Native OpenCrawling CLI (oc push)

# 1. Probe gRPC Push listener health and round-trip latency
$ oc push ping --host 127.0.0.1 --port 9096

# 2. Verify via Admin REST API Connector Checker
$ oc connector check --name Grpc_Push_Connector --type repository --config host=127.0.0.1 --config port=9096

# 3. Push an inline UPSERT document with Zero-Trust ACLs
$ oc push upsert \
  --job-id 1 \
  --id "https://ecm.example.com/contracts/2026-msa.md" \
  --text "# Master Services Agreement 2026..." \
  --acl-allow "GROUP_LEGAL,ROLE_EXECUTIVE" \
  --acl-deny "GROUP_CONTRACTORS" \
  --metadata "department=legal" \
  --token "my-secret-token"

# 4. Stream an entire directory of files over Bi-Directional HTTP/2 gRPC
$ oc push stream --bidi --job-id 1 --dir ./docs/contracts --token "my-secret-token"

# 5. Push a DELETE tombstone to purge vectors
$ oc push delete --job-id 1 --id "https://ecm.example.com/contracts/2026-msa.md" --token "my-secret-token"

2. Interactive Exploration with grpcurl (Server Reflection Enabled)

Because ProtoReflectionService is enabled by default, tools like grpcurl, Postman, and Kreya auto-discover the schema without needing local .proto files:

# Discover exposed gRPC services
$ grpcurl -plaintext 127.0.0.1:9096 list

# Push a document with Zero-Trust ACLs via grpcurl
$ grpcurl -plaintext \
  -H "authorization: Bearer my-secret-token" \
  -d '{
    "job_id": "1",
    "document": {
      "id": "https://ecm.example.com/contracts/2026-msa.md",
      "action": "UPSERT",
      "mime_type": "text/markdown",
      "text_content": "# Master Services Agreement\n\nGoverns enterprise cloud services...",
      "metadata": {
        "department": "legal",
        "contract_year": "2026"
      },
      "security": {
        "allow_sids": ["GROUP_LEGAL", "ROLE_EXECUTIVE"],
        "deny_sids": ["GROUP_CONTRACTORS"],
        "owner": "alice.smith",
        "tenant_id": "acme-corp"
      }
    },
    "options": {
      "immediate_flush": true,
      "routing_key": "legal-contracts"
    }
  }' \
  127.0.0.1:9096 opencrawling.push.v1.OisPushService/PushDocument

3. Embedding a Java 25 Bi-Directional Streaming Producer

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Metadata;
import io.grpc.stub.MetadataUtils;
import io.grpc.stub.StreamObserver;
import org.opencrawling.push.v1.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class GrpcBiDiStreamingProducer {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 9096)
                .usePlaintext()
                .build();

        Metadata headers = new Metadata();
        headers.put(Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER),
                "Bearer my-secret-token");

        OisPushServiceGrpc.OisPushServiceStub asyncStub = OisPushServiceGrpc.newStub(channel)
                .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers));

        CountDownLatch done = new CountDownLatch(1);
        StreamObserver<PushDocumentRequest> requestObserver = asyncStub.biDirectionalStream(
                new StreamObserver<>() {
                    @Override
                    public void onNext(IngestionAck ack) {
                        System.out.printf("ACK doc=%s status=%s at=%d%n",
                                ack.getDocumentId(), ack.getStatus(), ack.getAckTimestamp());
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                        done.countDown();
                    }

                    @Override
                    public void onCompleted() {
                        done.countDown();
                    }
                });

        for (int i = 1; i <= 5; i++) {
            requestObserver.onNext(PushDocumentRequest.newBuilder()
                    .setJobId("1")
                    .setDocument(OisDocumentPayload.newBuilder()
                            .setId("doc://kb/article-" + i)
                            .setAction(DocumentAction.UPSERT)
                            .setMimeType("text/plain")
                            .setTextContent("Real-time event-driven knowledge base entry #" + i)
                            .setSecurity(SecurityConfig.newBuilder()
                                    .addAllowSids("GROUP_ENGINEERING")
                                    .setTenantId("acme-corp")
                                    .build())
                            .build())
                    .build());
        }
        requestObserver.onCompleted();
        done.await(10, TimeUnit.SECONDS);
        channel.shutdownNow();
    }
}

8. Configuration Reference

Configure the gRPC Push Repository Connector via application.yml, environment variables, or the Admin UI Connector Registry (which includes a live Test gRPC Push Listener (Ping) diagnostic button):

9. Automated Verification & Decoupled Pipeline Testing

The gRPC Push Repository Connector is verified across both in-process unit/integration test suites and the full containerized decoupled stack: