Polling Reaktors

Overview

Polling reaktors are specialized reaktors that execute on a schedule and can signal whether more work exists through a boolean return value. This enables efficient batch processing and queue-based workflows where the reaktor continues processing until all work is completed.

When to Use Polling Reaktors

Use polling reaktors when you need to:

• Process queues or batches: Handle work items one at a time until the queue is empty
• Scheduled background tasks: Execute periodic tasks like data synchronization, cleanup, or monitoring
• Event-driven polling: Check for new events/messages and process them immediately
• Continuous processing: Process work as fast as possible when available, with periodic checks when idle

Comparison with Standard Reaktors

Aspect	Standard Reaktor	Polling Reaktor
Execution	On-demand (API call, chat message, email)	Scheduled (interval-based)
Return Value	Any type	Boolean (true = more work, false = wait)
Control	External triggers	REST API (start/stop/status)
State	Stateless	Persistent state (running/stopped)
Use Case	Request/response workflows	Background processing, queues

Registration API

import { registerPollingReaktor } from '@operaide/aktor';

registerPollingReaktor({
  reaktorDefinitionId: string,    // Unique identifier
  interval: string,                // '2m', '5s', '1h', '30ms', '1d'
  autoStart?: boolean,             // Default: false
  label?: string,                  // Optional UI label
  description?: string,            // Optional description
  createReaktor: () => Aktor<boolean>
});

Configuration Options

reaktorDefinitionId (required)

Unique identifier for this polling reaktor. Used in REST API endpoints and internal tracking.

reaktorDefinitionId: 'coffee-queue-processor'

interval (required)

Execution interval as a human-readable duration string:

• '30ms' - 30 milliseconds
• '5s' - 5 seconds
• '2m' - 2 minutes
• '1h' - 1 hour
• '1d' - 1 day

Note: There is a minimum interval of 3 seconds enforced by the distributed timer system for server coordination.

autoStart (optional, default: false)

Whether to start the reaktor automatically on server startup. Useful for critical background tasks that should always be running.

autoStart: true  // Starts on server init

label and description (optional)

Human-readable metadata displayed in UIs and documentation.

label: 'Coffee Queue Processor',
description: 'Processes coffee orders from the queue one at a time'

createReaktor (required)

Factory function that returns an Aktor<boolean>. The boolean value controls execution flow.

Boolean Return Contract

The reaktor's return value controls its execution behavior:

true - More Work Exists

• Reaktor executes again immediately (no interval wait)
• Continues until false is returned
• Enables fast batch processing

false - No Work Available

• Reaktor waits for the configured interval
• Next execution happens after interval expires
• Saves resources when queue is empty

Example Pattern

createReaktor() {
  return aktorIfElse({
    if_: aktorCheckQueue({}),        // Check if work exists
    then_: aktorProcessItem({}),     // Process & return true/false
    else_: aktorConst(false)         // No work, wait for interval
  });
}

Implementing Polling Reaktors

Pattern 1: Queue Processing

Process items from a queue until empty:

import {
  registerPollingReaktor,
  createAktorFunctionAsync,
  aktorIfElse,
  aktorConst
} from '@operaide/aktor';

// Check if queue has items
const aktorHasOrders = createAktorFunction('aktorHasOrders', hasOrders);
function hasOrders() {
  return queue.length > 0;
}

// Process one order (async operation)
const aktorProcessOrder = createAktorFunctionAsync('aktorProcessOrder', processOrder);
async function processOrder() {
  if (queue.length === 0) {
    return false;
  }

  const order = queue.shift();
  await processOrderAsync(order);  // Your async logic

  return queue.length > 0;  // More work?
}

registerPollingReaktor({
  reaktorDefinitionId: 'order-processor',
  interval: '5s',
  createReaktor() {
    return aktorIfElse({
      if_: aktorHasOrders({}),
      then_: aktorProcessOrder({}),  // Returns true if more orders
      else_: aktorConst(false)        // Queue empty, wait 5s
    });
  }
});

Pattern 2: Batch Processing with Settings

Use aktorSetting for configurable batch sizes:

import { aktorSetting } from '@operaide/aktor';
import { z } from 'zod';

const aktorBatchProcessor = createAktorFunctionAsync('aktorBatchProcessor', processBatch);
async function processBatch({ batchSize }: { batchSize: number }) {
  const items = await fetchNextBatch(batchSize);

  if (items.length === 0) {
    return false;  // No more items
  }

  await Promise.all(items.map(processItem));

  return items.length === batchSize;  // True if full batch (likely more items)
}

registerPollingReaktor({
  reaktorDefinitionId: 'batch-worker',
  interval: '1m',
  createReaktor() {
    const batchSize = aktorSetting(
      z.number().min(1).max(100).describe('Items per batch'),
      10
    );

    return aktorBatchProcessor({ batchSize });
  }
});

Pattern 3: Conditional Processing

Use aktorIfElse to branch based on conditions:

const aktorCheckAndProcess = createAktorFunctionAsync('aktorCheckAndProcess', checkAndProcess);
async function checkAndProcess() {
  const hasNewEmails = await checkForNewEmails();

  if (!hasNewEmails) {
    return false;  // No emails, wait for interval
  }

  const processed = await processEmails();
  return processed > 0;  // Continue if we processed emails
}

registerPollingReaktor({
  reaktorDefinitionId: 'email-processor',
  interval: '30s',
  createReaktor() {
    return aktorCheckAndProcess({});
  }
});

REST API Control

For each registered polling reaktor, the following endpoints are automatically generated:

Start a Polling Reaktor

POST /api/v1/polling/{reaktorId}/start

Example:

curl -X POST http://localhost:8811/api/v1/polling/coffee-queue-processor/start \
  -H "X-Operaide-Token: your-auth-token"

Response:

{
  "message": "Polling reaktor \"coffee-queue-processor\" started successfully"
}

Stop a Polling Reaktor

POST /api/v1/polling/{reaktorId}/stop

Example:

curl -X POST http://localhost:8811/api/v1/polling/coffee-queue-processor/stop \
  -H "X-Operaide-Token: your-auth-token"

Response:

{
  "message": "Polling reaktor \"coffee-queue-processor\" stopped successfully"
}

Get Reaktor Status

GET /api/v1/polling/{reaktorId}/status

Example:

curl http://localhost:8811/api/v1/polling/coffee-queue-processor/status \
  -H "X-Operaide-Token: your-auth-token"

Response:

{
  "state": "running",
  "interval": "5s",
  "lastRun": "2024-01-15T10:30:00Z",
  "nextRun": "2024-01-15T10:30:05Z"
}

State Persistence

Polling reaktor state is persisted in MongoDB and survives server restarts.

State Fields

• state: "running" or "stopped"
• interval: Configured interval string
• lastRun: ISO timestamp of last execution
• nextRun: ISO timestamp of next scheduled execution
• updatedAt: Last state change timestamp

Restart Behavior

With autoStart: true

• Reaktor automatically restarts if it was running before server restart
• Preserves the running state across deployments

With autoStart: false (default)

• Reaktor state is reset to "stopped" on server restart
• Must be manually started via REST API
• Prevents unexpected execution after deployment

This ensures non-critical polling reaktors don't automatically resume after server restart, giving operators control over when background processing begins.

Architecture and Implementation

Dependency Injection Pattern

Polling reaktors follow the same dependency injection pattern as registerReaktorDefinition:

@operaide/aktor (Pure TypeScript)
  ├─ PollingReaktor.ts
  │   └─ registerPollingReaktor() → calls _registerPollingReaktor
  └─ RegisterPollingReaktorPrivate.ts
      ├─ _registerPollingReaktor (no-op by default)
      └─ setRegisterPollingReaktorFunction() (for injection)

Meteor Extension (/imports/extensions/polling-reaktor)
  └─ registerInjector.ts
      └─ Injects Meteor implementation into @operaide/aktor

Benefits

• Clean Separation: @operaide/aktor has zero Meteor dependencies
• Testability: Can mock the implementation for testing
• Flexibility: Other frameworks can inject custom implementations
• Consistency: Same pattern as standard reaktors

Layered Architecture

1. Types Layer - TypeScript types and Zod schemas
2. Storage Layer - MongoDB collection for state persistence
3. Registry Layer - In-memory registration of polling reaktors
4. Execution Layer - Scheduling using distributed timers
5. API Layer - REST endpoints for control

Best Practices

1. Use Async Functions Correctly

Always use createAktorFunctionAsync for async functions:

// ✅ CORRECT
const aktorProcess = createAktorFunctionAsync('aktorProcess', processAsync);
async function processAsync() {
  await someAsyncOperation();
  return hasMoreWork;
}

// ❌ WRONG
const aktorProcess = createAktorFunction('aktorProcess', processAsync);

2. Handle Empty Queues Gracefully

Always check for empty queues before processing:

async function processOrder() {
  if (queue.length === 0) {
    return false;  // Stop processing
  }

  const order = queue.shift();
  await process(order);

  return queue.length > 0;  // Continue if more items
}

3. Set Appropriate Intervals

Choose intervals based on your use case:

• Fast processing: '5s' to '30s' - for responsive queue processing
• Periodic tasks: '5m' to '1h' - for scheduled maintenance
• Background sync: '1h' to '1d' - for low-priority sync tasks

4. Use autoStart Carefully

Only set autoStart: true for critical background tasks that must always run. For most cases, manual start via API provides better control.

5. Monitor Execution

Use the status endpoint to monitor reaktor health:

# Check status
curl http://localhost:8811/api/v1/polling/my-reaktor/status

# If stuck, restart
curl -X POST http://localhost:8811/api/v1/polling/my-reaktor/stop
curl -X POST http://localhost:8811/api/v1/polling/my-reaktor/start

6. Add Delays for Visibility (Demo Only)

For demo/testing purposes, add artificial delays to make processing visible:

async function processOrder() {
  const order = queue.shift();

  // Add delay for demo visibility (remove in production!)
  await new Promise(resolve => setTimeout(resolve, 2000));

  await actualProcessing(order);
  return queue.length > 0;
}

Advanced Patterns

Combining Multiple Aktors

Use aktorParallel to execute multiple checks:

import { aktorParallel } from '@operaide/aktor';

const aktorCheckMultipleSources = createAktorFunctionAsync(
  'aktorCheckMultipleSources',
  checkMultipleSources
);

async function checkMultipleSources({ sources }: { sources: string[] }) {
  const checks = await Promise.all(sources.map(checkSource));
  return checks.some(hasWork => hasWork);
}

createReaktor() {
  const sources = aktorConst(['email', 'api', 'database']);
  return aktorCheckMultipleSources({ sources });
}

Error Handling

Wrap processing in try-catch to prevent reaktor crashes:

async function processWithErrorHandling() {
  try {
    const item = queue.shift();
    await processItem(item);
    return queue.length > 0;
  } catch (error) {
    logger.error('Processing failed:', error);
    return false;  // Stop on error
  }
}

Rate Limiting

Implement rate limiting within the reaktor:

let lastProcessTime = 0;
const MIN_INTERVAL_MS = 1000;

async function processWithRateLimit() {
  const now = Date.now();
  const timeSinceLastProcess = now - lastProcessTime;

  if (timeSinceLastProcess < MIN_INTERVAL_MS) {
    await new Promise(resolve =>
      setTimeout(resolve, MIN_INTERVAL_MS - timeSinceLastProcess)
    );
  }

  lastProcessTime = Date.now();
  return await actualProcessing();
}

Troubleshooting

Reaktor Not Starting

1. Check server logs for errors
2. Verify reaktor is registered: GET /api/v1/polling
3. Check configuration syntax
4. Ensure createReaktor() returns Aktor<boolean>

Reaktor Stops Unexpectedly

1. Check for errors in reaktor execution
2. Verify async functions use createAktorFunctionAsync
3. Check if reaktor returned false (expected behavior)
4. Review server logs for exceptions

State Mismatch After Restart

If the UI shows "running" but reaktor isn't executing:

1. Server restart resets non-autoStart reaktors to "stopped"
2. Click "Stop" then "Start" to resynchronize
3. Or set autoStart: true if it should survive restarts

See Also

• @operaide/aktor Reference - Core library documentation
• op-demo-polling-coffee-queue - Complete working example with UI (see app's README)
• Building Reaktors - General reaktor development guide