Concurrent runs leads to duplicate rows

[simon-pactum]

When two concurrent sqlmesh plans are running, the latter only bails out at the end with something like:

Error: Plan 'e08afcaae9b3446fb09e98e9b38cdf31' is no longer valid for the target environment 'prod'. Expected previous plan ID: '72795b6460e14a9585aa46301749bc11', actual previous plan ID: '23fe369a90c34e789743abbba767791a'. Please recreate the plan and try again

But at that point it's already too late and duplicate rows have been created.

To reproduce:

sqlmesh: 0.158.2

# Config
gateways:
  postgres:
    connection:
      type: postgres
      host: localdb
      user: local
      password: local
      port: 5432
      database: local


default_gateway: postgres

model_defaults:
  dialect: postgres

MODEL (
  name sqlmesh_example.duplicate,
  kind FULL,
);

SELECT
  t.*
FROM
  -- Large enough to be "slow" enough for human input!
  -- Change after having created it once to force it to get recreated
  generate_series(1, 1000000) as t
  -- generate_series(1, 10000000) as t

1. sqlmesh plan say yes, let it do its thing
2. Change the model to force a rebuild, e.g. by adding an extra 0
3. Terminal A: sqlmesh plan wait for prompt, don't answer yet
4. Terminal B: sqlmesh plan wait for prompt, don't answer yet
5. Terminal A & B: Answer y in both terminals in quick succession

Now I have twice as many rows as expected in the prod environment:

# SELECT COUNT(1) FROM sqlmesh_example.duplicate;
  count   
----------
 20000000
(1 row)

This probably happened as both did the INSERT ... at the same time.

---

I'm not sure if this is an actual bug or just room for documentation improvement.

Either way, it would be nice if it bailed out with an error before causing issues

[simon-pactum]

One thing I'd appreciate more clarity on is if it's sufficient to limit concurrency per environment or globally, I would suspect globally as it's all the same physical tables, so two environments may interfere?

EDIT: Have to limit globally, can be reproduced as above but with sqlmesh plan foo/sqlmesh plan bar concurrently