level12 / pals Goto Github PK

View Code? Open in Web Editor NEW

44.0 44.0 9.0 67 KB

Easy distributed locking using PostgreSQL Advisory Locks.

License: Other

Python 100.00%

database locks postgresql python

pals's People

Contributors

Stargazers

Watchers

Forkers

ethanwillis absalon-james gward mayli lan17 battyone moser petrprikryl ominidx

pals's Issues

feature: default kwargs for locks

It would be neat to have a default_lock_params argument to Locker to allow things like blocking=False without needing to remember to do so for every .lock() call.

Thanks for creating this library, it works really really well.

Support transactional level advisory locks

@yudistrange asked in #29 (comment):

One thing that I found missing in django-pglocks was the support for transaction level advisory locks in postgres.
I couldn't find these here either.
Is there a plan to include these in the future?

There isn't a plan. I'm not necessarily opposed, it would just be a matter of calling a different function. What's the use case? Do you just want the "rollback" semantics so that the lock drops when the transaction drops?

Setup and deploy Sphinx Documentation

Look at Keg Storage as the example. Submit to Team B / NZ for review.

Use `lock_timeout` setting to expire blocking calls

see statement_timeout and lock_timeout: https://www.postgresql.org/docs/9.5/static/runtime-config-client.html

Bill thinks this would be a better approach than the non-blocking+loop method.

application_name is not set for each connection

It would be cool if we can use https://github.com/level12/pals/blob/master/pals/core.py#L51 as the connecition name, but it's not being used in the connection.

why use sqlalchemy instead of using driver directly?

I just read source, and looks like this lib just use sqlalchemy to execute sa.text sql, why use sqlalchemy instead of driver directly?

'acquire' is consistently misspelled

e.g. the exception AquireFailure and the keyword argument aquire_timeout are both spelled 'aquire' instead of 'acquire'. This makes the interface somewhat annoying to use.

get rid of pipenv, use pip tools

and also update readme

Prepare for SQLAlchemy 2.0

See https://docs.sqlalchemy.org/en/14/changelog/migration_20.html

psycopg2.InterfaceError: cursor already closed

When you try to create more locks than there are connections in the connection pool of the SqlA engine at the same time, the (N + 1)th lock will fail with the following error.

Traceback (most recent call last):
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/cursor.py", line 969, in fetchone
      row = dbapi_cursor.fetchone()
  psycopg2.InterfaceError: cursor already closed
  
  The above exception was the direct cause of the following exception:
  
  Traceback (most recent call last):
    File "/usr/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
      self.run()
    File "/usr/lib/python3.10/threading.py", line 953, in run
      self._target(*self._args, **self._kwargs)
    File "<...>/pals/pals/tests/test_core.py", line 220, in _critical_n
      with self.locker.lock(f"example-{n}"):
    File "<...>/pals/pals/core.py", line 156, in __exit__
      self.release()
    File "<...>/pals/pals/core.py", line 149, in release
      return result.scalar()
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/result.py", line 1276, in scalar
      return self._only_one_row(
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/result.py", line 559, in _only_one_row
      row = onerow(hard_close=True)
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/cursor.py", line 1816, in _fetchone_impl
      return self.cursor_strategy.fetchone(self, self.cursor, hard_close)
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/cursor.py", line 974, in fetchone
      self.handle_exception(result, dbapi_cursor, e)
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/cursor.py", line 955, in handle_exception
      result.connection._handle_dbapi_exception(
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2129, in _handle_dbapi_exception
      util.raise_(
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
      raise exception
    File "<...>/pals/.tox/py310-lowest/lib/python3.10/site-packages/sqlalchemy/engine/cursor.py", line 969, in fetchone
      row = dbapi_cursor.fetchone()
  sqlalchemy.exc.InterfaceError: (psycopg2.InterfaceError) cursor already closed
  (Background on this error at: https://sqlalche.me/e/14/rvf5)
  
    warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))

AttributeError in on_conn_checkin() (dbapi_connection can be None)

We just did a bit of maintenance on our PostgreSQL server. (It's actually an AWS RDS PostgreSQL, but I don't think that matters.)

This resulted in the following error:

[2020-03-02 16:01:38,519 E 14 fn.flasknoir.errorhandlers] POST /heartbeat?: internal error handling request
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1246, in _execute_context
    cursor, statement, parameters, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 581, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.AdminShutdown: terminating connection due to administrator command
SSL connection has been closed unexpectedly


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1476, in _handle_dbapi_exception
    util.raise_from_cause(sqlalchemy_exception, exc_info)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 398, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 152, in reraise
    raise value.with_traceback(tb)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1246, in _execute_context
    cursor, statement, parameters, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 581, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.OperationalError: (psycopg2.errors.AdminShutdown) terminating connection due to administrator command
SSL connection has been closed unexpectedly

Nothing surprising there, and nothing to do with pals. Just giving some context.

Then it turns out that pals' on_conn_checkin() method blew up while unwinding the stack from that exception:

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1949, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.7/site-packages/flask/app.py", line 1935, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  [...my application code...]
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 982, in execute
    return meth(self, multiparams, params)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py", line 287, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1101, in _execute_clauseelement
    distilled_params,
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1250, in _execute_context
    e, statement, parameters, cursor, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1487, in _handle_dbapi_exception
    self.engine.pool._invalidate(dbapi_conn_wrapper, e)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 325, in _invalidate
    connection.invalidate(exception)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 956, in invalidate
    self._checkin()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 847, in _checkin
    fairy=self,
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 697, in _finalize_fairy
    connection_record.checkin()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 527, in checkin
    pool.dispatch.checkin(connection, self)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/event/attr.py", line 322, in __call__
    fn(*args, **kw)
  File "/usr/local/lib/python3.7/site-packages/pals/core.py", line 45, in on_conn_checkin
    with dbapi_connection.cursor() as cur:
AttributeError: 'NoneType' object has no attribute 'cursor'

Conclusion: it is possible for SQLalchemy to pass None as dbapi_connection. pals should defend against this -- probably by doing nothing!

Very minor bug: this request still would have failed with a 500 error.

Allow acquire timeout to be specified like for threading.Lock

threading.Lock.aquire takes an argument named timeout, using seconds rather than milliseconds with -1 to indicate no timeout.

It would be nice to have a drop-in-compatible interface for pals.Lock.aquire, using something like

def aquire(self, ..., timeout=None):
    if aquire_timeout is None and timeout is not None:
        acquire_timeout = 0 if timeout<0 else 1+int(timeout*1000)

Also, in the same function: It looks like it is not currently possible to override the default aquire_timeout with 0 (no timeout) due to the acquire_timeout or ... construction. It should perhaps be changed to

    acquire_timeout = acquire_timeout if acquire_timeout is not None else self.aquire_timeout

Support nested locks for the same lock on the same connection

PG advisory locks support nested locks. So, the same connection can request the same lock multiple times. We don't currently support that, but should consider it.

Needs update to license field in setup.py

Use SPDX classifier (https://spdx.org/licenses/) for license field in setup.py

Update bleach

Opened since our policy is to track time against the issue, not the pr, see https://github.com/level12/pals/pulls

Adjust default lock timeout from 1s to 30s.

Seems like a more reasonable number to me.

When raising AcquireFailure provide more info for debugging

For example, the name of the lock and why it's failed to acquire.

Intermittent IdleInTransactionSessionTimeout

Thanks for this useful package. Intermittently, my applications encounter the following exception:

IdleInTransactionSessionTimeout: terminating connection due to idle-in-transaction timeout
SSL connection has been closed unexpectedly

  File "sqlalchemy/engine/base.py", line 1245, in _execute_context
    self.dialect.do_execute(
  File "sqlalchemy/engine/default.py", line 588, in do_execute
    cursor.execute(statement, parameters)
  File "ddtrace/contrib/dbapi/__init__.py", line 99, in execute
    return self._trace_method(self.__wrapped__.execute, self._self_datadog_name, query, {}, query, *args, **kwargs)
  File "ddtrace/contrib/psycopg/patch.py", line 56, in _trace_method
    return super(Psycopg2TracedCursor, self)._trace_method(method, name, resource, extra_tags, *args, **kwargs)
  File "ddtrace/contrib/dbapi/__init__.py", line 69, in _trace_method
    return method(*args, **kwargs)
InternalError: (psycopg2.errors.IdleInTransactionSessionTimeout) terminating connection due to idle-in-transaction timeout
SSL connection has been closed unexpectedly

[SQL: set lock_timeout = %(timeout)s]
[parameters: {'timeout': 10000}]
(Background on this error at: http://sqlalche.me/e/2j85)
  File "flask/app.py", line 2446, in wsgi_app
    response = self.full_dispatch_request()
  File "flask/app.py", line 1951, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "ddtrace/contrib/flask/helpers.py", line 22, in wrapper
    return func(pin, wrapped, instance, args, kwargs)
  File "ddtrace/contrib/flask/helpers.py", line 31, in wrapper
    return wrapped(*args, **kwargs)
  File "flask/app.py", line 1820, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "flask/_compat.py", line 39, in reraise
    raise value
  File "flask/app.py", line 1949, in full_dispatch_request
    rv = self.dispatch_request()
  File "ddtrace/contrib/flask/helpers.py", line 22, in wrapper
    return func(pin, wrapped, instance, args, kwargs)
  File "ddtrace/contrib/flask/patch.py", line 474, in _traced_request
    return wrapped(*args, **kwargs)
  File "flask/app.py", line 1935, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "ddtrace/contrib/flask/wrappers.py", line 25, in trace_func
    return wrapped(*args, **kwargs)
  File "connexion/decorators/decorator.py", line 48, in wrapper
    response = function(request)
  File "connexion/decorators/uri_parsing.py", line 144, in wrapper
    response = function(request)
  File "connexion/decorators/validation.py", line 384, in wrapper
    return function(request)
  File "connexion/decorators/parameter.py", line 121, in wrapper
    return function(**kwargs)
  File "src/api.py", line 179, in get_premises
    "location_risks": _get_risks(id, cached_only)['risks'],
  File "src/api.py", line 52, in _get_risks
    with lock:
  File "pals/core.py", line 132, in __enter__
    if not self.acquire():
  File "pals/core.py", line 101, in acquire
    self.conn.execute(timeout_sql, timeout=acquire_timeout)
  File "sqlalchemy/engine/base.py", line 982, in execute
    return meth(self, multiparams, params)
  File "sqlalchemy/sql/elements.py", line 293, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "sqlalchemy/engine/base.py", line 1095, in _execute_clauseelement
    ret = self._execute_context(
  File "sqlalchemy/engine/base.py", line 1249, in _execute_context
    self._handle_dbapi_exception(
  File "sqlalchemy/engine/base.py", line 1476, in _handle_dbapi_exception
    util.raise_from_cause(sqlalchemy_exception, exc_info)
  File "sqlalchemy/util/compat.py", line 398, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File "sqlalchemy/util/compat.py", line 152, in reraise
    raise value.with_traceback(tb)
  File "sqlalchemy/engine/base.py", line 1245, in _execute_context
    self.dialect.do_execute(
  File "sqlalchemy/engine/default.py", line 588, in do_execute
    cursor.execute(statement, parameters)
  File "ddtrace/contrib/dbapi/__init__.py", line 99, in execute
    return self._trace_method(self.__wrapped__.execute, self._self_datadog_name, query, {}, query, *args, **kwargs)
  File "ddtrace/contrib/psycopg/patch.py", line 56, in _trace_method
    return super(Psycopg2TracedCursor, self)._trace_method(method, name, resource, extra_tags, *args, **kwargs)
  File "ddtrace/contrib/dbapi/__init__.py", line 69, in _trace_method
    return method(*args, **kwargs)

Are connections being left in the "idle in transaction" state? Can these transactions be committed or rolled back?

Update readme code example

The readme code example says:

Recommended usage pattern

Then

But more recommended and easier is to use the lock as a context manager

But then gives a third example showing the context manager with exception handling which is really what you should be doing. So probably need to clean the wording and examples up a bit.

Not really sure what, if anything, PALs should do about this, maybe just some documentation around the problem would be helpful.