The contract

Between sink and the first value there is a handshake, and almost everything surprising about Combine is explained by it.

1. Subscribe. The subscriber is handed to the publisher via receive(subscriber:).
2. Subscription. The publisher creates a Subscription — the per-subscriber state — and hands it back with receive(subscription:).
3. Demand. The subscriber calls subscription.request(_:) with how many values it wants. Until it does, nothing is sent.
4. Values, then completion. The publisher sends up to that many values, each returning more demand, and eventually one .finished or .failure.

protocol Subscriber<Input, Failure> {
    func receive(subscription: Subscription)
    func receive(_ input: Input) -> Subscribers.Demand
    func receive(completion: Subscribers.Completion<Failure>)
}

Note the return type of the middle method. Every value delivered is an opportunity to ask for more, which is how demand accumulates without a separate channel.

Writing one, once

Nobody writes custom subscribers in production. Writing one once, in a playground, is still the fastest way to stop finding Combine mysterious:

final class OneAtATime: Subscriber {
    typealias Input = Int
    typealias Failure = Never

    func receive(subscription: Subscription) {
        subscription.request(.max(1))          // ask for exactly one
    }

    func receive(_ input: Int) -> Subscribers.Demand {
        print("got \(input)")
        return .max(1)                         // having consumed one, ask for one more
    }

    func receive(completion: Subscribers.Completion<Never>) {
        print("done: \(completion)")
    }
}

(1...5).publisher.subscribe(OneAtATime())

Change .max(1) to .none in receive(_:) and the stream delivers exactly one value and then stops forever — no completion, no error, just silence. That is not a bug; that is the contract working. The publisher is not permitted to send a second value because nobody asked for one.

What sink and assign really are

Both are Subscribers.Sink and Subscribers.Assign under a convenience method, and both request .unlimited immediately. That is why you never think about demand in normal use — and why a fast publisher into a slow sink buffers without limit rather than slowing down.

assign(to:on:) writes each value to a key path on an object:

model.$name
    .assign(to: \.text, on: label)
    .store(in: &cancellables)

Cancellable and what cancelling does

sink and assign return an AnyCancellable. It has one job: on deinit, it calls cancel(), which tears down the subscription and releases the chain.

This is why the stored Set<AnyCancellable> exists, and why forgetting .store(in:) produces a pipeline that runs for zero values — the cancellable is deallocated at the end of the function, and the subscription with it.

Cancellation propagates upstream: cancel at the sink and every operator above it is torn down, up to and including a URLSession request, which is genuinely cancelled rather than merely ignored.

The type explosion, and eraseToAnyPublisher

Every operator wraps its upstream in a new generic type, so a five-operator chain has a type name like:

Publishers.RemoveDuplicates<Publishers.Debounce<Publishers.CompactMap<…>, RunLoop>>

You cannot write that in an API, so erase it at the boundary:

func searchResults(for query: String) -> AnyPublisher<[Result], Never> {
    api.search(query)
        .replaceError(with: [])
        .receive(on: DispatchQueue.main)
        .eraseToAnyPublisher()
}

Erase at the edge of a type, not between every operator. AnyPublisher boxes the chain and adds a layer of indirection per value — negligible once, wasteful if you do it four times in one pipeline.