MCPcopy Create free account
hub / github.com/PostHog/duckgres / messageLoop

Method messageLoop

server/conn.go:1118–1192  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

1116 return err
1117 }
1118
1119 // Upgrade connection to TLS
1120 tlsConn := tls.Server(c.conn, c.server.tlsConfig)
1121 if err := tlsConn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil {
1122 return fmt.Errorf("failed to set TLS deadline: %w", err)
1123 }
1124 if err := tlsConn.Handshake(); err != nil {
1125 return fmt.Errorf("TLS handshake failed: %w", err)
1126 }
1127 if err := tlsConn.SetDeadline(time.Time{}); err != nil {
1128 return fmt.Errorf("failed to clear TLS deadline: %w", err)
1129 }
1130
1131 // Replace connection with TLS connection
1132 c.conn = tlsConn
1133 c.reader = bufio.NewReader(tlsConn)
1134 c.writer = bufio.NewWriter(tlsConn)
1135 tlsUpgraded = true
1136
1137 c.logger().Info("TLS connection established.", "remote_addr", c.conn.RemoteAddr())
1138 continue
1139 }
1140
1141 // Handle cancel request
1142 if startup.CancelRequest {
1143 if startup.CancelCredentialsPresent {
1144 c.server.CancelQuery(BackendKey{Pid: startup.CancelPID, SecretKey: startup.CancelSecretKey})
1145 }
1146 return errCancelHandled
1147 }
1148 params := startup.Params
1149
1150 // Reject non-TLS connections
1151 if !tlsUpgraded {
1152 c.sendError("FATAL", "28000", "SSL/TLS connection required. Connect with sslmode=require or higher.")
1153 return fmt.Errorf("client did not request SSL")
1154 }
1155
1156 c.username = params["user"]
1157 c.database = params["database"]
1158 c.applicationName = params["application_name"]
1159
1160 // Honor `-c duckgres.query_source=...` / `-c duckgres.s3_cache=...`
1161 // startup options (libpq `options` keyword / PGOPTIONS). Other GUCs in
1162 // `options` are not applied here. An invalid value rejects the
1163 // connection with FATAL 22023 — the same treatment
1164 // resolveWorkerProfile gives an invalid duckgres.worker_* startup
1165 // option, and what PostgreSQL itself does with an invalid `options`
1166 // GUC value. (In standalone mode there is no cache proxy, so
1167 // duckgres.s3_cache only records session state here; the worker-swap
1168 // path is control-plane-only.)
1169 if opts := ParseStartupOptions(params["options"]); len(opts) > 0 {
1170 if v, ok := opts[ClientIdleTimeoutGUCName]; ok {
1171 if err := c.applyStartupIdleTimeout(v); err != nil {
1172 c.sendError("FATAL", "22023", err.Error())
1173 return fmt.Errorf("invalid %s startup option", ClientIdleTimeoutGUCName)
1174 }
1175 }

Callers 5

serveMethod · 0.95
runChildWorkerFunction · 0.95
runPipelineFunction · 0.80
RunMessageLoopFunction · 0.80

Calls 10

armIdleReadDeadlineMethod · 0.95
loggerMethod · 0.95
handleQueryMethod · 0.95
writeReadyForQueryMethod · 0.95
flushWriterMethod · 0.95
ReadMessageFunction · 0.92
isConnectionBrokenFunction · 0.85
TimeoutMethod · 0.80
ErrorMethod · 0.45

Tested by 2

runPipelineFunction · 0.64