ConcurrencyMacros

ConcurrencyMacros is a production-focused Swift Concurrency macro package for the patterns teams implement repeatedly: lock-backed shared state (with practical checked Sendable adoption), in-flight deduplication, callback-to-stream bridging, timeouts, retries, and bounded concurrent collection work.

The package keeps macro call sites small while routing behavior through explicit runtime helpers with documented safety constraints.

Requirements

• Swift 6.2
• iOS 17+
• macOS 14+
• tvOS 17+
• watchOS 10+

Installation

Add the package dependency in Package.swift:

dependencies: [
    .package(url: "https://github.com/naykutguven/ConcurrencyMacros.git", from: "0.1.0")
]

Add the library product to your target:

.target(
    name: "MyApp",
    dependencies: [
        .product(name: "ConcurrencyMacros", package: "ConcurrencyMacros")
    ]
)

Quick Start

Start with these flagship macros in most apps:

• @ThreadSafe: lock-backed mutable state with practical checked Sendable adoption for final classes.
• @SingleFlightActor: deduplicate in-flight actor work by key.
• #withTimeout: enforce a hard deadline for async operations.
• #retrying: recover from transient failures with explicit retry policy.
• #concurrentMap: run bounded concurrent fan-out while preserving input order.

import ConcurrencyMacros
import Foundation

struct Avatar: Sendable {
    let data: Data
}

protocol AvatarAPI: Sendable {
    func fetchAvatar(for userID: UUID) async throws -> Avatar
}

@ThreadSafe
final class AvatarCache: Sendable {
    var values: [UUID: Avatar] = [:]
}

actor AvatarService {
    private let api: AvatarAPI
    private let cache = AvatarCache()

    init(api: AvatarAPI) {
        self.api = api
    }

    @SingleFlightActor(key: { (userID: UUID) in userID })
    func avatar(for userID: UUID) async throws -> Avatar {
        if let cached = cache.values[userID] {
            return cached
        }

        let fetched = try await #withTimeout(.seconds(5)) {
            try await #retrying(
                max: 2,
                backoff: .exponential(initial: .milliseconds(200), multiplier: 2, maxDelay: .seconds(2)),
                jitter: .full
            ) {
                try await api.fetchAvatar(for: userID)
            }
        }

        cache.values[userID] = fetched
        return fetched
    }
}

func loadAvatars(userIDs: [UUID], service: AvatarService) async throws -> [Avatar] {
    try await #concurrentMap(userIDs, limit: .fixed(4)) { id in
        try await service.avatar(for: id)
    }
}

Optional: Stream Bridging Path

If you integrate callback-first SDKs, add @StreamBridge as a companion flagship macro:

import ConcurrencyMacros

final class PriceFeedClient: Sendable {
    @StreamBridge(
        as: "priceStream",
        event: .label("handler"),
        cancel: .ownerMethod("stopObserving"),
        buffering: .bufferingNewest(32),
        safety: .strict
    )
    func observePrice(
        symbol: String,
        handler: @escaping @Sendable (PriceTick) -> Void
    ) -> ObservationToken {
        sdk.observePrice(symbol: symbol, handler: handler)
    }

    func stopObserving(_ token: ObservationToken) {}
}

func consume(client: PriceFeedClient) async {
    for await tick in client.priceStream(symbol: "AAPL") {
        print(tick)
    }
}

Macro Index

| Macro | Kind | Purpose | Applies To | | --- | --- | --- | --- | | @ThreadSafe | Attached (member, memberAttribute) | Synthesizes lock-backed state and rewrites mutable stored properties | Class declarations | | @ThreadSafeInitializer | Attached (body) | Helper rewrite for initializer assignment staging | Initializers (helper/support) | | @ThreadSafeProperty | Attached (accessor) | Helper rewrite for lock-backed property accessors | Mutable stored properties (helper/support) | | @SingleFlightActor | Attached (body, peer) | Deduplicates in-flight actor method work by key | Actor instance methods | | @SingleFlightClass | Attached (body, peer) | Deduplicates in-flight class method work by key | final class instance methods | | @StreamBridge | Attached (body, peer) | Generates AsyncStream / AsyncThrowingStream wrappers from callback registration methods | Actor/class instance methods | | @StreamBridgeDefaults | Attached (member) | Declares default stream-bridge options for a nominal type | Nominal types (helper/support) | | @StreamToken | Attached (extension) | Synthesizes StreamBridgeTokenCancellable conformance | Class/struct/enum tokens (helper/support) | | #withTimeout | Freestanding expression | Runs an async operation with timeout cancellation | Expressions | | #retrying | Freestanding expression | Retries async throwing work with backoff and jitter | Expressions | | #concurrentMap | Freestanding expression | Concurrent async map with stable output order | Expressions | | #concurrentCompactMap | Freestanding expression | Concurrent async compact-map with stable output order | Expressions | | #concurrentFlatMap | Freestanding expression | Concurrent async flat-map with stable outer ordering | Expressions | | #concurrentForEach | Freestanding expression | Concurrent async side-effect execution | Expressions |

@ThreadSafe

What it does

@ThreadSafe synthesizes lock-backed internal state and redirects mutable stored-property access through generated accessors. It also makes adopting checked Sendable on stateful classes more practical by centralizing mutable state behind a synchronized, Sendable internal model.

When to use

Use it when you need synchronous read/write APIs on shared mutable class state while preserving consistent lock semantics.

<details open><summary>Example</summary>

import ConcurrencyMacros

@ThreadSafe
final class SessionStore {
    var sessionsByID: [String: Session] = [:]
    var activeUserID: String?

    func upsert(_ session: Session) {
        sessionsByID[session.id] = session
    }
}

</details>

Safety notes

• Intended for class declarations.
• When a class has no initializer, each mutable stored property must have a default value.
• Rewriting is applied to mutable stored properties and designated initializers; convenience initializers are not rewritten.
• The generated state container is lock-backed and Sendable.

@SingleFlightActor

What it does

@SingleFlightActor rewrites an actor instance method so concurrent calls with the same key share one in-flight operation.

When to use

Use it for expensive actor-isolated async operations where duplicate concurrent requests should coalesce.

<details open><summary>Example</summary>

import ConcurrencyMacros

actor ProfileService {
    @SingleFlightActor(key: { (userID: Int) in userID })
    func profile(userID: Int) async throws -> Profile {
        try await api.fetchProfile(id: userID)
    }
}

</details>

Safety notes

• Deduplication is in-flight only; results are not cached after completion.
• Currently supported only on nominal actor instance methods (not extensions, static, class, or nonisolated methods).
• Method must be async; typed throws, generic methods, opaque some returns, and unsupported parameter forms (for example inout) are rejected.
• key: is required and cannot be a string literal.
• using: is optional, but if provided it must reference an existing store value (identifier/member access), not key paths or call expressions.
• Generated wrappers enforce Sendable for the evaluated key and forwarded parameters.

@SingleFlightClass

What it does

@SingleFlightClass rewrites a class instance method so concurrent calls with the same key share one in-flight operation via an explicit store.

When to use

Use it when request coalescing is needed in reference-type services that cannot be actors.

<details open><summary>Example</summary>

import ConcurrencyMacros

final class ProfileService: Sendable {
    private static let sharedFlights = ThrowingSingleFlightStore<Profile>()

    @SingleFlightClass(key: { (userID: Int) in userID }, using: Self.sharedFlights)
    func profile(userID: Int) async throws -> Profile {
        try await api.fetchProfile(id: userID)
    }
}

</details>

Safety notes

• Deduplication is in-flight only; results are not cached after completion.
• using: is required and must reference an existing store value (identifier/member access).
• Currently supported only on nominal class instance methods (not extensions, static, or class methods).
• Enclosing class must be final and explicitly conform to checked Sendable; @unchecked Sendable is rejected.
• Method must be async; typed throws, generic methods, opaque some returns, and unsupported parameter forms are rejected.
• Generated wrappers enforce Sendable for self, evaluated key, and forwarded parameters.

@StreamBridge

What it does

@StreamBridge generates a stream-returning wrapper from a callback registration method, producing AsyncStream or AsyncThrowingStream based on selected callbacks.

When to use

Use it when bridging callback-based SDK observation APIs to structured async stream consumption.

<details open><summary>Example</summary>

import ConcurrencyMacros

final class PriceFeedClient: Sendable {
    @StreamBridge(
        as: "priceStream",
        event: .label("handler"),
        cancel: .ownerMethod("stopObserving"),
        buffering: .bufferingNewest(32),
        safety: .strict
    )
    func observePrice(
        symbol: Strin