fix(multipart): Wrap items in Managed

[elramen]

Wrap multipart items with Managed so that outcomes are emitted when something goes wrong (i.e. when items get dropped).

Fixes https://linear.app/getsentry/issue/RELAY-231/use-managed-in-multipart

[elramen]

@jjbayer @Dav1dde The PR has taken your input into account, namely to neither use accept directly from endpoints, add items to a non-managed envelope, nor add more methods to Managed<Box<Envelope>>. By fulfilling these requirements, however, the following lines are repeated in three endpoints now:

let envelope = Envelope::from_request(Some(path.event_id), meta);
let mut envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());
let mut has_event = false;
for item in items {
    envelope.merge_with(item, |envelope, item, records| {
        if !has_event && item.creates_event() {
            records.modify_by(DataCategory::Error, 1);
            has_event = true;
        }
        envelope.add_item(item);
    });
}

I personally find one of the following options less verbose and easier to understand:

let mut envelope = Envelope::from_request(Some(path.event_id), meta);
for item in items {
    item.accept(|i| envelope.add_item(i));
}
let envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());

or

let mut envelope = Envelope::from_request(Some(path.event_id), meta);
envelope.add_managed_items(items);
let envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());

...

impl Envelope {
    pub fn add_managed_items(&mut self, items: ManagedItems) {
        for item in items {
            item.accept(|i| envelope.add_item(i));
        }
    }
}

or

let envelope = Envelope::from_request(Some(path.event_id), meta);
let mut envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());
envelope.add_managed_items(items);

...

impl Managed<Box<Envelope>> {
    pub fn add_managed_items(&mut self, items: ManagedItems) {
        let mut has_event = false;
        for item in items {
            self.merge_with(item, |envelope, item, records| {
                if !has_event && item.creates_event() {
                    records.modify_by(DataCategory::Error, 1);
                    has_event = true;
                }
                envelope.add_item(item);
            });
        }
    }
}

or

let envelope = Envelope::from_request(Some(path.event_id), meta);
let envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone()).with_managed_items(items);

...

impl Managed<Box<Envelope>> {
    pub fn with_managed_items(self, items: ManagedItems) -> Self {
        let meta = self.meta.clone();
        let mut envelope = self.accept(|e| e);
        for item in items {
            item.accept(|item| envelope.add_item(item));
        }
        Managed::from_parts(envelope, meta)
    }

If you prefer to keep things the way they are in this PR, I'm happy to disagree and commit!

[elramen]

Several places with new - old. Would it be fine to just use lenient in these cases?

[jjbayer]

I think so, but it might make sense to debug_assert that the new size is >= old size.

[jjbayer]

By fulfilling these requirements, however, the following lines are repeated in three endpoints now:

You could also call records.lenient(DataCategory::Error); in the merge_with closure, but that's just a sloppier version of your current solution.

I would prefer if a Managed<Item> which contains a minidump would return a quantity in the error category, and then we manually close the gaps where this is violated (e.g. when a minidump and a placeholder event together represent 1 error). But this should be in a separate PR, so for this one I'm fine with any of the solutions that you listed.

Draft PR for the accounting change: #5919

[linear-code]

RELAY-231 Use Managed in multipart

[Dav1dde]

I think the necessity for:

    envelope.merge_with(item, |envelope, item, records| {
        if !has_event && item.creates_event() {
            records.modify_by(DataCategory::Error, 1);
            has_event = true;
        }
        envelope.add_item(item);
    });

highlights a complexity (maybe a flaw?) in our accounting logic. That an item in an envelope has a different meaning than standalone.

So we need a solution to build up an envelope from individual items. I think hiding the accounting special case behind a function/facade makes sense.

This is a bit of unexplored territory in Relay, so far we've only really taken things out of a Managed<Box<Envelope>> and never directly constructed it. It's likely we'd run into similar problems elsewhere.
A lot of the API also assumes this usage, for example the way Rejected<> has been built, if you have multiple Managed<> instances at once, this becomes significantly less useful, it cannot guarantee all individual items have actually been rejected.

My initial idea was to only go Managed<> once the entire envelope has been assembled. You have a good use-case now why you'd want to go Managed<> maybe before having the final envelope.

To be honest none of the solutions or things I can think of make me entirely happy. The thing I probably want to do least is add of this to the Managed<> interface as the problem is with the accounting of envelopes not really with the Managed<>. impl Managed<Box<Envelope>> maybe is a decent place but I really tried to get rid of it.

Maybe we need something different which allows us to build up managed instances piece by piece.

---

Long story, I think what I'd go with here just to get this unblocked is one of these:

1. Hide the construction of the envelope in a utility function and use it only for the endpoints:

let envelope = Envelope::from_request(Some(path.event_id), meta);
let mut envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());
let mut has_event = false;
for item in items {
    envelope.merge_with(item, |envelope, item, records| {
        if !has_event && item.creates_event() {
            records.modify_by(DataCategory::Error, 1);
            has_event = true;
        }
        envelope.add_item(item);
    });
}

This entire thing, hidden behind a facade.

2. Which may work better as an approach (but I haven't actually tried):

Instead of having Vec<Managed<Item>>, try Managed<Vec<Item>>, this is much closer to the original intended usage.
To construct the envelope, it's now just a items.map(move |items, records| records.modify_by(DataCategory::Error, 1); Envelope::from_request(); ...). Which we can also potentially hide behind a function.