flowctl: Batch draft spec upserts, rather than incorrectly attempting to paginate the response

[jshearer]

Description:

When pagination was originally introduced, this code-path was incorrectly switched to use it, despite it actually being an upsert which doesn't support pagination

flow/crates/flowctl/src/draft/author.rs

Lines 118 to 124 in f618926

	let rows: Vec<SpecSummaryItem> = api_exec_paginated(
	client
	.from("draft_specs")
	.select("catalog_name,spec_type")
	.upsert(String::from_utf8(body).expect("serialized JSON is always UTF-8"))
	.on_conflict("draft_id,catalog_name"),
	)

Requests that don't support pagination just straight up ignore the range header. That wasn't a problem because the pagination logic only tried to fetch the next page if it got back exactly the number of rows in its requested range:

flow/crates/flowctl/src/pagination.rs

Line 124 in f618926

if resp.len() == request.page_size

So, this was an extremely rare edge case that would only get hit if you tried to publish exactly 1000 specs. But, I noticed that for requests that do respect pagination, sometimes postgrest will actually return one more row than was requested, so I updated this to a >= in #1516:

flow/crates/flowctl/src/pagination.rs

Lines 124 to 126 in 8c90141

	// Sometimes, it seems, we can get back more than the requested page size.
	// So far I've only seen this on a request of 1,000 and a response of 1,001.
	if resp.len() >= request.page_size

Before that change is committed, any request that returns more rows than were requested will incorrectly cause pagination to think it has reached the end and result in an incomplete result set being returned. After that change is committed, any attempt to publish >= 1000 specs will sit in an infinite loop attempting to load the next page of a request that doesn't support pagination.

This PR fixes both cases:

• It switches pagination to use >= check to correctly load paginated data where PostgREST returns more rows than the requested range
• It switches catalog publishes to batch their inputs and not use pagination when making upsert queries.

Testing:

Figuring out what was going on here was a bit complicated as the reason I thought my original PR fixed the issue was that I was working ontop of my branch in #1516 that had the pagination fix applied, but not the fix for flowctl catalog publish. So, after confirming that my locally-built flowctl is just ontop of this branch jshearer/batch_drafts, which is ontop of master:

$ flowctl catalog pull-specs --captures=true --collections=true --flat --target flow.yaml --prefix test/many_collections
Wrote 1734 specifications under file:///Users/js/Documents/estuary/test/flow.yaml.

$ flowctl catalog publish --source flow.yaml
Created draft: 0e...
Will publish the following 1734 specs:
╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┬─────────────────╮
│ Name                                                                                                                            │ Type            │
╞═════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╪═════════════════╡
│ test/many_collections/...                                                                                                       │ capture         │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼─────────────────┤

---

Also reverts --name back to its previous behavior since pagination doesn't make sense there.

---

This change is 

[jshearer]

I had wanted to use buffered_unordered() to limit the request concurrency, but got some compile errors that I haven't seen before and wasn't sure what to do with them, so I used FuturesUnordered like we do elsewhere, presumably for the same reason:

error: implementation of `Send` is not general enough
  --> crates/flowctl/src/main.rs:22:18
   |
22 |     let handle = runtime.spawn(async move { cli.run().await });
   |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ implementation of `Send` is not general enough
   |
   = note: `Send` would have to be implemented for the type `&flowctl::auth::Auth`
   = note: ...but `Send` is actually implemented for the type `&'0 flowctl::auth::Auth`, for some specific lifetime `'0`

error: implementation of `Send` is not general enough
  --> crates/flowctl/src/main.rs:22:18
   |
22 |     let handle = runtime.spawn(async move { cli.run().await });
   |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ implementation of `Send` is not general enough
   |
   = note: `Send` would have to be implemented for the type `&Cli`
   = note: ...but `Send` is actually implemented for the type `&'0 Cli`, for some specific lifetime `'0`

[jgraettinger]

LGTM % a couple comments