Twelve-Factor Examples, Factors IX through XII

These examples finish the shared order-service lifecycle. They focus on replacement, verification, operational evidence, and one-off work. Choose a language in any tab group. Every group on this page follows that selection.

Factor IX: Disposability

The web process starts quickly, advertises readiness only after its dependencies are usable, and removes readiness before shutdown drains active work. The worker stops accepting jobs before its grace period expires.

TypeScript

Node signal handlers enter one idempotent shutdown path. The web process stops dispatch, drains readiness, closes Fastify and attached clients, and uses the configured grace deadline.

  const stop = onceAsync(async () => {
    try {
      clearInterval(interval);
      await drain(readiness, async () => {
        await app.close();
        await queue.close();
        await store.close();
        await tracing?.shutdown();
      }, config.SHUTDOWN_GRACE_MS);
    } catch {
      emit(eventRecord("web", config.RELEASE_ID, "error", "web.shutdown_deadline", {
        errorCategory: "shutdown_deadline"
      }));
      process.exitCode = 1;
    }
  });
  process.once("SIGTERM", () => void stop());
  process.once("SIGINT", () => void stop());

Factor IX: Disposability TypeScript source

Python

Dramatiq consumers pause before the worker drains. The configured deadline bounds the pause and stop sequence, then the broker and store close through the shared readiness wrapper.

def shutdown_worker(
    worker: DrainingWorker,
    readiness: Readiness,
    broker: Closeable,
    store: Closeable,
    grace_ms: int,
) -> None:
    def close() -> None:
        for consumer in worker.consumers.values():
            consumer.pause()
        for consumer in worker.consumers.values():
            if not consumer.paused_event.wait(grace_ms / 1000):
                raise TimeoutError("consumer did not stop accepting work")
        worker.stop(timeout=grace_ms)
        broker.close()
        store.close()

    drain(readiness, close, grace_ms)

Factor IX: Disposability Python source

Go

The web command waits for cancellation, drops readiness, emits a draining event, and calls http.Server.Shutdown with a fixed timeout.

  case <-ctx.Done():
  case err := <-errorsCh:
    if !errors.Is(err, http.ErrServerClosed) {
      return err
    }
  }
  state.BeginDrain()
  telemetry.Event(context.Background(), logger, "web", cfg.ReleaseID, "web.draining")
  shutdownCtx, cancel := context.WithTimeout(context.Background(), time.Duration(cfg.ShutdownGraceMS)*time.Millisecond)
  defer cancel()
  if err := server.Shutdown(shutdownCtx); err != nil {
    return fmt.Errorf("web shutdown deadline: %w", err)
  }
  telemetry.Event(context.Background(), logger, "web", cfg.ReleaseID, "web.stopped")
  return nil

Factor IX: Disposability Go source

Evidence to collect: Send the termination signal during active HTTP and worker tasks. Verify readiness fails first, accepted work finishes or retries safely, and the process exits within the configured grace period.

Factor X: Dev/prod parity

Parity focuses on production-significant behavior. The local and CI paths use the same database and queue service types as production while keeping credentials, scale, and topology local.

TypeScript

Vitest starts a disposable PostgreSQL service through Testcontainers. The test uses the production pg client and migrations, so transaction and query behavior are not replaced by an in-memory fake.

  beforeAll(async () => {
    [postgres, redis] = await Promise.all([
      new PostgreSqlContainer("postgres:18-alpine").withDatabase("orders").withUsername("orders").withPassword("local_only").start(),
      new GenericContainer("redis:8-alpine").withExposedPorts(6379).start()
    ]);
    databaseUrl = postgres.getConnectionUri();
    redisUrl = `redis://${redis.getHost()}:${redis.getMappedPort(6379)}`;
  }, 120_000);

  afterAll(async () => {
    await Promise.all([postgres?.stop(), redis?.stop()]);
  });

Factor X: Dev/prod parity TypeScript source

Python

pytest starts the same PostgreSQL service type used in production. Psycopg executes the schema and assertions through its normal wire protocol.

def services() -> tuple[PostgresContainer, DockerContainer, str, str]:
    postgres = PostgresContainer("postgres:18-alpine", username="orders", password="local_only", dbname="orders")
    redis = DockerContainer("redis:8-alpine").with_exposed_ports(6379)
    postgres.start()
    redis.start()
    redis_url = f"redis://{redis.get_container_host_ip()}:{redis.get_exposed_port(6379)}"
    return postgres, redis, postgres.get_connection_url(driver=None), redis_url

Factor X: Dev/prod parity Python source

Go

Go’s testing package starts PostgreSQL through the Testcontainers module. pgx applies the schema and exercises the same protocol as the application.

func TestPostgresAndRedisWorkflow(t *testing.T) {
  ctx := context.Background()
  postgresContainer, err := postgres.Run(ctx, "postgres:18-alpine", postgres.WithDatabase("orders"), postgres.WithUsername("orders"), postgres.WithPassword("local_only"), postgres.BasicWaitStrategies())
  if err != nil {
    t.Fatal(err)
  }
  t.Cleanup(func() { _ = postgresContainer.Terminate(context.Background()) })
  databaseURL, err := postgresContainer.ConnectionString(ctx, "sslmode=disable")
  if err != nil {
    t.Fatal(err)
  }
  pool, err := store.NewPool(ctx, databaseURL)
  if err != nil {
    t.Fatal(err)
  }
  t.Cleanup(pool.Close)
  if err := migrations.Apply(ctx, pool, "001"); err != nil {
    t.Fatal(err)
  }
  if err := migrations.Ready(ctx, pool); err != nil {
    t.Fatal(err)
  }

Factor X: Dev/prod parity Go source

Evidence to collect: List every production-significant difference, name the test layer that covers it, and fail review when a difference has no owner or check.

Factor XI: Logs

Application processes write structured events to standard output and standard error. They do not choose log files, retention periods, or remote storage destinations.

TypeScript

The event builder reads the active OpenTelemetry span and adds its trace identifier when present. The emitter serializes the structured record to standard output.

export function eventRecord(
  processType: ProcessType,
  releaseId: string,
  level: Level,
  event: string,
  fields: Partial<Pick<EventFields, "orderId" | "jobId" | "attempt" | "errorCategory">> = {},
  activeContext: Context = context.active()
): EventFields {
  const traceId = trace.getSpan(activeContext)?.spanContext().traceId;
  return {
    timestamp: new Date().toISOString(),
    level,
    event,
    service: "twelve-factor-orders",
    processType,
    releaseId,
    ...(traceId ? { traceId } : {}),
    ...fields
  };
}

export function emit(record: EventFields): void {
  process.stdout.write(`${JSON.stringify(record)}\n`);
}

Factor XI: Logs TypeScript source

Python

structlog renders JSON to standard output. OpenTelemetry contributes the current trace identifier when a valid span is active.

def event_record(
    process_type: ProcessType,
    release_id: str,
    level: str,
    event: str,
    **fields: Any,
) -> dict[str, Any]:
    span_context = trace.get_current_span().get_span_context()
    record: dict[str, Any] = {
        "timestamp": datetime.now(UTC).isoformat(),
        "level": level,
        "event": event,
        "service": "twelve-factor-orders",
        "processType": process_type,
        "releaseId": release_id,
        **fields,
    }
    if span_context.is_valid:
        record["traceId"] = format(span_context.trace_id, "032x")
    return record


def emit(record: dict[str, Any]) -> None:
    values = dict(record)
    event = str(values.pop("event"))
    LOGGER.msg(event, **values)

Factor XI: Logs Python source

Go

slog writes JSON to standard output. OpenTelemetry extracts trace context from the request context and adds it to the same event.

func Event(ctx context.Context, logger *slog.Logger, processType, releaseID, event string, attrs ...any) {
  fields := []any{"service", "twelve-factor-orders", "processType", processType, "releaseId", releaseID, "event", event}
  span := oteltrace.SpanContextFromContext(ctx)
  if span.IsValid() {
    fields = append(fields, "traceId", span.TraceID().String())
  }
  logger.InfoContext(ctx, event, append(fields, attrs...)...)
}

Factor XI: Logs Go source

Evidence to collect: Submit one failing request and reconstruct its path from the external event stream using release, request, and trace identifiers.

Factor XII: Admin processes

The migration command uses the same application packages, dependency graph, configuration loader, and artifact as web and worker processes. The platform runs it as a bounded job with an attributable result.

TypeScript

Commander parses a narrow migration command. The action reuses the validated config, PostgreSQL client, and migration function shipped with web and worker processes.

export async function runAdmin(argv: string[]): Promise<void> {
  const config = parseConfig(process.env, "admin");
  const pool = createPool(config.DATABASE_URL);
  const command = buildAdminCommand(async (target) => {
    emit(eventRecord("admin", config.RELEASE_ID, "info", "admin.migration_started"));
    await migrate(pool, target);
    emit(eventRecord("admin", config.RELEASE_ID, "info", "admin.migration_finished"));
  });
  try {
    await command.parseAsync(argv, { from: "user" });
  } finally {
    await pool.end();
  }
}

Factor XII: Admin processes TypeScript source

Python

Typer exposes one versioned operation. The command imports the same settings, connection factory, and migration code as the rest of the installed package.

def migrate_command(target: str) -> None:
    settings = load_settings(os.environ, "admin")
    pool = create_pool(settings.database_url)
    try:
        pool.open(wait=True)
        emit(event_record("admin", settings.release_id, "info", "admin.migration_started"))
        run_database_migration(pool, target)
        emit(event_record("admin", settings.release_id, "info", "admin.migration_finished"))
    finally:
        pool.close()


def build_cli(run: Callable[[str], None] = migrate_command) -> typer.Typer:
    cli = typer.Typer(add_completion=False)

    @cli.callback()
    def root() -> None:
        """Run bounded administrative commands."""

    @cli.command("migrate")
    def migration(target: str = typer.Option(..., help="Exact migration target")) -> None:
        try:
            run(target)
        except ValueError as error:
            raise typer.BadParameter(str(error), param_hint="--target") from None

    return cli

Factor XII: Admin processes Python source

Go

Cobra adds a migration subcommand to the normal binary. It loads the shared config and pgxpool resource, then delegates schema work to the versioned migration package.

func adminCommand(logger *slog.Logger) *cobra.Command {
  admin := &cobra.Command{Use: "admin", Short: "Run one-off administrative processes"}
  var target string
  migrate := &cobra.Command{Use: "migrate", Short: "Apply a bounded schema migration", RunE: func(cmd *cobra.Command, _ []string) error {
    if target != "001" {
      return fmt.Errorf("unsupported migration target")
    }
    cfg, err := config.Load("admin")
    if err != nil {
      return err
    }
    shutdownTracing, err := telemetry.Configure(cfg.TelemetryMode)
    if err != nil {
      return err
    }
    defer shutdownTracing(context.Background())
    ctx := context.Background()
    pool, err := store.NewPool(ctx, cfg.DatabaseURL)
    if err != nil {
      return fmt.Errorf("database unavailable")
    }
    defer pool.Close()
    telemetry.Event(ctx, logger, "admin", cfg.ReleaseID, "admin.migration_started", "target", target)
    if err := migrations.Apply(ctx, pool, target); err != nil {
      return err
    }
    telemetry.Event(ctx, logger, "admin", cfg.ReleaseID, "admin.migration_finished", "target", target)
    return nil
  }}
  migrate.Flags().StringVar(&target, "target", "", "migration target")
  _ = migrate.MarkFlagRequired("target")
  admin.AddCommand(migrate)
  return admin
}

Factor XII: Admin processes Go source

Evidence to collect: Tie the migration to an actor, command, release digest, start time, end time, and result without opening an unrestricted shell in a running instance.

Reference releases

The examples on this page are exact excerpts from independently runnable releases:

• TypeScript v1.0.2, verified with npm run verify
• Python v1.0.2, verified with uv run python scripts/check_twelve_factor.py
• Go v1.0.3, verified with ./scripts/check-twelve-factor.sh

Navigation

• Return to the Twelve-Factor overview
• Back to Factor V: Build, release, run through Factor VIII: Concurrency
• Use the Twelve-Factor service audit
• Kubernetes