Postgres memleak caused by prepared statements growing forever when prepareCache is overflowing

[silles79]

Our Postgres db keeps running out of RAM.

We have an infinity stream of events from Kafka, each we groupBy(number = 1000) then go to the db and fetch in batch

We have a few queries which are parametrized with lists like

sql """ select ...
where id in (${SomeCodecs.id.list(ids.size)});
"""

We have a session with a pool, prepareCacheSize = 1024
We have about 25 queries

I think what happens is that we have more prepared statements ( lets say about 25 x 1000 ) than the cache size ,
and when the local cache runs out of space it just disposed it from the local cache but not from the service with deallocate.

When the same query with same number of params to list comes up, skunk just prepares again since it's no longer in the cache, so the number prepared statement prepared statement on the postgres server just keeps growing until it either dies or we restart the app

There are a few thing I could/will do at our side but I thought it make sense to report this as a bug

[matthughes]

@silles79 Is there someway you can confirm this on the Postgres side of things? I suspect something similar is happening but was hoping to see evidence of it in pg_prepared_statements but the table is empty.

[matthughes]

@silles79 Ok, so only the connection that created the prepared statements can see what's in that table for them.

I modified my application to wake up and print out the contents of that table every 10 seconds and even with these values set to 10:

commandCache:  Int = 1024,
queryCache:    Int = 1024,
parseCache:    Int = 1024,

I'm seeing the count in pg_prepared_statements to only go up over time up and above my limit. Same behavior even with it set to 0.

[matthughes]

Need to work on a better minimized reproduction but this is what I do:

• with a single session db pool
• run a unique in query 2000 times (e.g., List(1), List(1,2), List(1,2,3), etc)
• then check the pg_prepared_statements table

Even though the previous transaction has been completed and we aren't keeping the prepared statements around, they are all still there in the next transaction. It prints out 2000 prepared statements from the previous transaction.

      // def inQuery(names: List[String]): Query[names.type, String] =
      //   sql"select state from pg_stat_activity where state IN (${text.values.list(names)})".query(text)

      // def psStatementCount: Query[Void, (Int, String, String, Instant)] =
      //   sql"select pg_backend_pid(), name, statement, prepare_time from pg_prepared_statements order by prepare_time"
      //     .query(
      //       int4 *: text *: text *: offsetDtInstantCodec
      _ <- Stream.eval {
        withTxFromSessionPool(pool) { (s, tx) =>
          1.to(2000).toList.traverse { x =>
            val values = 1.to(x).toList.map(_.toString)
            multiview.MultiviewRepository.inQuery(values)(s, tx).void
          }
        }
      }
      pgStatementDump = Stream.awakeEvery[IO](10.seconds).evalMap { _ =>
        withTxFromSessionPool(pool) { (s, tx) =>
          multiview.MultiviewRepository.getPreparedStatementCount(s, tx).map { entries =>
            println(s"COUNT IS ${entries.size}")
            entries.take(10).foreach { case (pid, name, statement, time) =>
              println(s"   ${pid} $time $name ${statement.take(100)}")
            }
          }
        }
      }

[matthughes]

With debug on, I see Skunk sending the close message:

server  ← ReadyForQuery(Active)
server  → Parse(statement_3,select state from pg_stat_activity where state IN (($1)),List(25))
server  → Describe(S, statement_3)
server  → Flush
server  ← ParseComplete
server  ← ParameterDescription(List(25))
server  ← RowDescription(Field(state, 25))
server  → Bind(portal_4,statement_3,List(Some(1)))
server  → Flush
server  ← BindComplete
server  → Execute(portal_4,2147483647)
server  → Flush
server  ← CommandComplete(Select(0))
server  → Sync
server  ← ReadyForQuery(Active)
server  → Close(P,portal_4)
server  → Flush
server  ← CloseComplete

But then after that transaction has closed, we still see the statement:

server    3127 2024-12-16T01:56:50.168376Z statement_3 select state from pg_stat_activity where state IN (($1))

[matthughes]

Ok, here's a full reproduction:

Run a Postgres container. We're going to first customize the config file to show debug5 so we can see as much of the message protocol as possible:

> docker run -i --rm postgres cat /usr/share/postgresql/postgresql.conf.sample > my-postgres.conf
> echo "log_min_messages = debug5" >> my-postgres.conf
> docker run -v "$PWD/my-postgres.conf":/etc/postgresql/postgresql.conf -p5432:5432 --name some-postgres -e POSTGRES_PASSWORD=mysecretpassword -d postgres -c 'config_file=/etc/postgresql/postgresql.conf'
> docker logs -f some-postgres

In a separate window, run the attached test case. In two separate sessions, we issue a prepared statement, and then query the prepared statements table:

//> using dep org.tpolecat::skunk-core:1.0.0-M8

import cats.effect.IOApp
import cats.effect.ExitCode
import cats.effect.IO
import cats.syntax.all.*
import skunk.*
import skunk.codec.all.*
import skunk.implicits.*
import cats.effect.kernel.Resource
import java.time.OffsetDateTime
import org.typelevel.otel4s.trace.Tracer.Implicits.noop

object SkunkMain extends IOApp {

  def inQuery(names: List[String]): Query[names.type, String] =
    sql"select state from pg_stat_activity where state IN (${text.values.list(names)})".query(text)

  val psPreparedStatements: Query[Void, (Int, String, String, OffsetDateTime)] =
    sql"select pg_backend_pid(), name, statement, prepare_time from pg_prepared_statements order by prepare_time"
      .query(
        int4 *: text *: text *: timestamptz
      )

  private def databaseSessionPool: Resource[IO, Resource[IO, Session[IO]]] =
    Session
      .pooled[IO](
        host = "localhost",
        port = 5432,
        user = "postgres",
        database = "postgres",
        password = "mysecretpassword".some,
        max = 1,
        commandCache = 0,
        queryCache = 0,
        parseCache = 0,
        debug = true
      )

  def run(args: List[String]): IO[ExitCode] =
    val activeStates = List("active")
    databaseSessionPool
      .use { sessions =>
        sessions.use { session =>
          IO.println("Preparing and using query") >>
            session.execute(inQuery(activeStates))(activeStates)
        } >>
          sessions.use { session =>
            session.execute(psPreparedStatements).flatMap { results =>
              IO.println(s"Prepared Statement Count: ${results.size}") >>
                results.traverse(IO.println)
            }
          }
      }
      .as(ExitCode.Success)
}

You'll see that Skunk sends the Close("P") command for portal_2

Preparing and using query
 → Parse(statement_1,select state from pg_stat_activity where state IN (($1)),List(25))
 → Describe(S, statement_1)
 → Flush
 ← ParseComplete
 ← ParameterDescription(List(25))
 ← RowDescription(Field(state, 25))
 → Bind(portal_2,statement_1,List(Some(active)))
 → Flush
 ← BindComplete
 → Execute(portal_2,2147483647)
 → Flush
 ← RowData(List(Some(active)))
 ← CommandComplete(Select(1))
 → Sync
 ← ReadyForQuery(Idle)
 → Close(P,portal_2)
 → Flush
 ← CloseComplete
 → Query(UNLISTEN *)
 ← CommandComplete(Unlisten)
 ← ReadyForQuery(Idle)
 → Query(RESET ALL)
 ← CommandComplete(Reset)
 ← ReadyForQuery(Idle)
 → Query(select pg_backend_pid(), name, statement, prepare_time from pg_prepared_statements order by prepare_time)
 ← RowDescription(Field(pg_backend_pid, 23); Field(name, 25); Field(statement, 25); Field(prepare_time, 1184))
 ← RowData(List(Some(78), Some(statement_1), Some(select state from pg_stat_activity where state IN (($1))), Some(2024-12-16 13:32:11.270206+00)))
 ← CommandComplete(Select(1))
 ← ReadyForQuery(Idle)

On the postgres side, you can see the bind to portal_2 occurring but there is no corresponding close message:

2024-12-16 13:43:40.409 GMT [79] DEBUG:  00000: parse statement_1: select state from pg_stat_activity where state IN (($1))
2024-12-16 13:43:40.409 GMT [79] LOCATION:  exec_parse_message, postgres.c:1422
2024-12-16 13:43:40.409 GMT [79] DEBUG:  00000: StartTransaction(1) name: unnamed; blockState: DEFAULT; state: INPROGRESS, xid/subid/cid: 0/1/0
2024-12-16 13:43:40.409 GMT [79] LOCATION:  ShowTransactionStateRec, xact.c:5626
2024-12-16 13:43:40.426 GMT [79] DEBUG:  00000: bind portal_2 to statement_1
2024-12-16 13:43:40.426 GMT [79] LOCATION:  exec_bind_message, postgres.c:1657
2024-12-16 13:43:40.443 GMT [79] DEBUG:  00000: CommitTransaction(1) name: unnamed; blockState: STARTED; state: INPROGRESS, xid/subid/cid: 0/1/0
2024-12-16 13:43:40.443 GMT [79] LOCATION:  ShowTransactionStateRec, xact.c:5626
2024-12-16 13:43:41.478 GMT [79] DEBUG:  00000: StartTransaction(1) name: unnamed; blockState: DEFAULT; state: INPROGRESS, xid/subid/cid: 0/1/0
2024-12-16 13:43:41.478 GMT [79] LOCATION:  ShowTransactionStateRec, xact.c:5626
2024-12-16 13:43:41.478 GMT [79] DEBUG:  00000: CommitTransaction(1) name: unnamed; blockState: STARTED; state: INPROGRESS, xid/subid/cid: 0/1/0
2024-12-16 13:43:41.478 GMT [79] LOCATION:  ShowTransactionStateRec, xact.c:5626
2024-12-16 13:43:41.493 GMT [79] DEBUG:  00000: StartTransaction(1) name: unnamed; blockState: DEFAULT; state: INPROGRESS, xid/subid/cid: 0/1/0
2024-12-16 13:43:41.493 GMT [79] LOCATION:  ShowTransactionStateRec, xact.c:5626
2024-12-16 13:43:41.493 GMT [79] DEBUG:  00000: CommitTransaction(1) name: unnamed; blockState: STARTED; state: INPROGRESS, xid/subid/cid: 0/1/0
2024-12-16 13:43:41.493 GMT [79] LOCATION:  ShowTransactionStateRec, xact.c:5626

Not evidence of much but interesting that I see one and not the other. I took a Wireshark capture and confirmed Skunk's protocol. So unless someone else has some ideas, I'm going to reach out to the postgres mailing list.

[matthughes]

skunk_pg.pcapng.zip

Here is Wireshark capture of the exchange

[matthughes]

Ok I think I have found the issue. We are closing portals (Close('P')) but not statements (Close('S')). Portal = prepared statement that is ready to execute.

We only actually clean up statements when Protocol.cleanup is called and that is only called when the session is really closed.

        override def cleanup: F[Unit] =
          parseCache.value.values.flatMap(xs => xs.traverse_(protocol.Close[F].apply))

There is a further bug which is exasperated if you set query/commandCache to low values. If you set the cache to 10 but create 100 prepared statements in your session, even calling cleanup won't solve this problem as the code above only knows about the first 10 items in the cache. I don't know that this is really an issue as when the connection is closed, Postgres will clear this memory.

[matthughes]

I guess part of what I don't understand is how are folks using prepared statements in their code? In addition to tons of our queries being dynamically generated leading to lots of unique statements, you also have the problem with IN queries:

select * from foo where id IN (?);
select * from foo where id IN (?, ?);

These are two different prepared statements. Not saying my app doesn't have any stable queries but the dynamic ones far outweigh the stable ones.

I see a couple possible fixes:

• Create opt-out execute methods in Session: Session.executeNoCache that clears prepared statement at statement end
• If queryCache/commandCache are set to 0, we should always be cleaning these up
• Expose cleanup to Session

[silles79]

As a workaround I replaced the list codecs:

where id in (${SomeCodecs.id.list(ids.size)});
which created up to ids.size prepared statements,

with

where ccbp.id = ANY(${intKeyArrayCodec[ Product.Id ]})

intKeyArrayCodec is just a wrapper and using Codecs.array.
This creates only 1 prepared statement.

This reduced by many folds our postgres mem usage.

[silles79]

Ok I think I have found the issue. We are closing portals (Close('P')) but not statements (Close('S')). Portal = prepared statement that is ready to execute.

We only actually clean up statements when Protocol.cleanup is called and that is only called when the session is really closed.

        override def cleanup: F[Unit] =
          parseCache.value.values.flatMap(xs => xs.traverse_(protocol.Close[F].apply))

There is a further bug which is exasperated if you set query/commandCache to low values. If you set the cache to 10 but create 100 prepared statements in your session, even calling cleanup won't solve this problem as the code above only knows about the first 10 items in the cache. I don't know that this is really an issue as when the connection is closed, Postgres will clear this memory.

I can confirm: yes when the connection is closed the server releases the sessions prepared statements.

My issue is that we never close the connections (unless we deploy a new app), because we have a infinity stream.

It takes a few days and postgress db runs out of RAM.

[matthughes]

intKeyArrayCodec is just a wrapper and using Codecs.array.
This creates only 1 prepared statement.

I was looking at that too. Was curious if there were any performance drawbacks to that but looks like Postgres rewrites the IN statement to an array one anyway:

localhost:5436 db@db=# explain select * from stream where id IN (1,2,3);
                                    QUERY PLAN
----------------------------------------------------------------------------------
 Bitmap Heap Scan on stream  (cost=12.89..24.56 rows=3 width=487)
   Recheck Cond: (id = ANY ('{1,2,3}'::bigint[]))
   ->  Bitmap Index Scan on stream_pkey  (cost=0.00..12.89 rows=3 width=0)
         Index Cond: (id = ANY ('{1,2,3}'::bigint[]))
(4 rows)

Time: 12.994 ms
localhost:5436 db@db=# explain select * from stream where id = ANY(ARRAY[1,2,3]);
                                    QUERY PLAN
----------------------------------------------------------------------------------
 Bitmap Heap Scan on stream  (cost=12.89..24.56 rows=3 width=487)
   Recheck Cond: (id = ANY ('{1,2,3}'::integer[]))
   ->  Bitmap Index Scan on stream_pkey  (cost=0.00..12.89 rows=3 width=0)
         Index Cond: (id = ANY ('{1,2,3}'::integer[]))
(4 rows)