SIGSEGV using `journal.Reader`, thread safety issue?

Start by reproducing the supplied Python threading example and tracing journal.Reader.wait() and Reader.close() through the _reader extension shown in the traceback. Compare the concurrent wait/close behavior with the safe close-before-join sequence; done means the reproducible scenario no longer segfaults and the behavior is covered by a regression check.

Env:

$ python3 --version
Python 3.11.2
$ systemctl --version
systemd 252 (252.36-1~deb12u1)
+PAM +AUDIT +SELINUX +APPARMOR +IMA +SMACK +SECCOMP +GCRYPT -GNUTLS +OPENSSL +ACL +BLKID +CURL +ELFUTILS +FIDO2 +IDN2 -IDN +IPTC +KMOD +LIBCRYPTSETUP +LIBFDISK +PCRE2 -PWQUALITY +P11KIT +QRENCODE +TPM2 +BZIP2 +LZ4 +XZ +ZLIB +ZSTD -BPF_FRAMEWORK -XKBCOMMON +UTMP +SYSVINIT default-hierarchy=unified
$ python3 -c "from systemd import journal; print(journal.__version__)"
235

Issue:

I got sporadically a segfault using journal.Reader from python-systemd module.
Thereby it seemed to happen by stop of thread, monitoring journal with Reader.

I could reproduce it also under dgb, here is the callstack:

[New Thread 0x7ffff5f876c0 (LWP 89339)]

Thread 2 "f2b/f.DummyJail" received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0x7ffff5f876c0 (LWP 89339)]
0x00007ffff6f86d63 in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
(gdb) where
#0  0x00007ffff6f86d63 in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
#1  0x00007ffff6f877f5 in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
#2  0x00007ffff6f6a848 in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
#3  0x00007ffff6f99502 in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
#4  0x00007ffff6f69403 in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
#5  0x00007ffff6f6961e in ?? () from /lib/x86_64-linux-gnu/libsystemd.so.0
#6  0x00007ffff6f6d11a in sd_journal_get_fd () from /lib/x86_64-linux-gnu/libsystemd.so.0
#7  0x00007ffff6f6f3c9 in sd_journal_wait () from /lib/x86_64-linux-gnu/libsystemd.so.0
#8  0x00007ffff6cadf2c in ?? () from /usr/lib/python3/dist-packages/systemd/_reader.cpython-311-x86_64-linux-gnu.so
#9  0x0000000000548ac8 in ?? ()
#10 0x00000000005188b3 in _PyObject_MakeTpCall ()
#11 0x000000000052c2f0 in _PyEval_EvalFrameDefault ()
#12 0x0000000000585e54 in ?? ()
#13 0x0000000000584e98 in ?? ()
#14 0x000000000053034e in _PyEval_EvalFrameDefault ()
#15 0x0000000000585e54 in ?? ()
#16 0x0000000000584e98 in ?? ()
#17 0x000000000067b32c in ?? ()
#18 0x0000000000656384 in ?? ()
#19 0x00007ffff7d381f5 in start_thread (arg=<optimized out>) at ./nptl/pthread_create.c:442
#20 0x00007ffff7db889c in clone3 () at ../sysdeps/unix/sysv/linux/x86_64/clone3.S:81

It can be either some missing lock or atomic handling somewhere in journal.Reader, because if I rewrote the code to avoid too earlier clean-up, closing the journal with Reader.close() before join of thread (now it ensures that the thread working with Reader is really stopped, before clean-up happens), it doesn't happen anymore.
So my assumption is if TH2 is somewhere within reader.wait() and TH1 would invoke reader.close(), it could segfault.
Sure, it can be avoided on the user side, just SF is SF and shall not happen normally.

Naive attempts to reproduce it with simple wait/close in different threads did not work, but I could find a bit complex scenario (with a busy-wait & lock using to synchronize the timing a bit), where it is pretty reproducible, almost every time not later than on 10 iteration:

import time
from systemd import journal
from threading import Thread, Lock
l = Lock(); # just used for "better" timing here to get SF

def test():
  stat = {};
  jr = journal.Reader()
  def mon():
    print('start')
    try:
      jr.seek_head()
      print('work')
      stat['started'] = 1
      l.acquire(); l.release()
      # *-* [A], thread is here as another thread is on *-* [B]
      while (j := jr):
        if j.wait(3) in (journal.NOP, journal.APPEND):
          j.get_next()
    except Exception as e:
      print('ERR: %s' % e)
    if jr: jr.close()
  th = Thread(target=mon); 
  th.start()
  # just "sync" wait and close calls:
  while not stat: time.sleep(0.00001);
  l.acquire(); l.release()
  # *-* [B], thread is here as another thread is on *-* [A]
  # do evil close:
  jr.close(); jr = None
  th.join()
  print('OK')

for i in range(100): # just 100 attempts, because of sporadic matter
  test()