bitflicker64 commented on code in PR #824: URL: https://github.com/apache/hugegraph-toolchain/pull/824#discussion_r4303681932
########## hugegraph-hubble/hubble-fe/src/api/text2gqlStream.js: ########## @@ -0,0 +1,198 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +import i18n from '../i18n'; +import {withLanguageHeader} from './languageHeader'; +import {handleRequestAuthStatus} from './request'; +import {sanitizePublicError} from '../utils/publicError'; + +const MAX_STREAM_BYTES = 1000000; +const MAX_QUERY_LENGTH = 10000; + +const streamError = (message, status) => { + handleRequestAuthStatus(status); + const error = new Error(sanitizePublicError(message, i18n.t('analysis.query.stream_incomplete'))); + error.response = {status, data: {status, message: error.message}}; + return error; +}; + +// Parse SSE incrementally. Neither an unvalidated delta nor EOF is a result. +const readQueryStream = async (body, onDelta) => { + if (!body?.getReader) { + throw streamError(); + } + const reader = body.getReader(); + const decoder = new TextDecoder('utf-8', {fatal: true}); + let buffer = ''; + let data = []; + let received = 0; + let draftLength = 0; + let result; + const dispatch = () => { + if (!data.length) { + return; + } + let event; + try { + event = JSON.parse(data.join('\n')); + } + catch { + throw streamError(); + } + data = []; + if (event.type === 'error') { + throw streamError(event.message, event.status); + } + if (event.type === 'delta') { + if (typeof event.text !== 'string' || (draftLength += event.text.length) > MAX_QUERY_LENGTH) { + throw streamError(); + } + if (event.text) { + onDelta?.(event.text); + } + } + else if (event.type === 'result') { + if (typeof event.query !== 'string' || event.query.length > MAX_QUERY_LENGTH + || typeof event.valid !== 'boolean') { + throw streamError(); + } + result = {status: 200, data: event}; + } + }; + const line = value => { + if (!value) { + dispatch(); + } + else if (value.startsWith('data:')) { + data.push(value.slice(5).replace(/^ /, '')); + } + }; + const consume = final => { + let index; + while ((index = buffer.search(/[\r\n]/)) !== -1) { + if (!final && buffer[index] === '\r' && index === buffer.length - 1) { + break; + } + const separator = buffer[index] === '\r' && buffer[index + 1] === '\n' ? 2 : 1; + line(buffer.slice(0, index)); + buffer = buffer.slice(index + separator); + if (result) { + return; + } + } + if (final && !result) { + line(buffer); + buffer = ''; + dispatch(); + } + }; + try { + while (!result) { + const {done, value} = await reader.read(); + if (done) { + buffer += decoder.decode(); + consume(true); + break; + } + received += value.byteLength; + if (received > MAX_STREAM_BYTES) { + throw streamError(); + } + buffer += decoder.decode(value, {stream: true}); + consume(false); + } + if (!result) { + throw streamError(); + } + return result; + } + finally { + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } +}; + +const readErrorBody = async body => { + if (!body?.getReader) { + return undefined; + } + const reader = body.getReader(); + const decoder = new TextDecoder(); + let text = ''; + let bytes = 0; + try { + while (bytes <= 32000) { + const {done, value} = await reader.read(); + if (done) { + return JSON.parse(text + decoder.decode()); + } + bytes += value.byteLength; + if (bytes > 32000) { + return undefined; + } + text += decoder.decode(value, {stream: true}); + } + } + catch { + return undefined; + } + finally { + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } +}; + +export const streamText2Gql = async (graphspace, graph, params, {signal, onDelta} = {}) => { + const controller = new AbortController(); + let timedOut = false; + const abort = () => controller.abort(); + signal?.addEventListener('abort', abort, {once: true}); + if (signal?.aborted) { + abort(); + } + const timer = setTimeout(() => { + timedOut = true; + controller.abort(); + }, 65000); + try { + const response = await fetch(`/api/v1.3/graphspaces/${encodeURIComponent(graphspace)}` + + `/graphs/${encodeURIComponent(graph)}/text2gql/stream`, { + method: 'POST', credentials: 'same-origin', signal: controller.signal, + headers: withLanguageHeader({'Content-Type': 'application/json', Accept: 'text/event-stream'}), Review Comment: Important: Errors raised before the stream starts reach the browser as HTTP 500 with an empty body. This request sends only `Accept: text/event-stream` and `Text2GqlController.java:70` declares `produces = text/event-stream`, so `ExceptionAdvisor` cannot write its JSON `Response`. That covers an expired session from `LoginInterceptor`, the graph access check, an unavailable shared model and every check in `prepare`. On embedded Tomcat with the real `LoginInterceptor` and `ExceptionAdvisor`, both a missing session and an `ExternalException` return 500 with no body. `readErrorBody` finds nothing, so a 401 no longer redirects to login as the old axios path did. The test "expires a business 401 returned as JSON before the stream is established" mocks a reply the backend never sends. With `Accept: text/event-stream, application/json` the same requests return 401 JSON and the advisor's JSON `Response`. Please widen Accept, or send these as an SSE `error` event, and test through the advisor. ########## hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/query/Text2GqlController.java: ########## @@ -0,0 +1,207 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hugegraph.controller.query; + +import java.io.IOException; +import java.util.LinkedHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import java.util.Map; + +import org.apache.hugegraph.common.Constant; +import org.apache.hugegraph.controller.BaseController; +import org.apache.hugegraph.driver.HugeClient; +import org.apache.hugegraph.entity.query.Text2GqlRequest; +import org.apache.hugegraph.entity.query.Text2GqlResult; +import org.apache.hugegraph.exception.ExternalException; +import org.apache.hugegraph.handler.MessageSourceHandler; +import org.apache.hugegraph.service.query.Text2GqlSettingsService; +import org.apache.hugegraph.service.query.Text2GqlService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.MediaType; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequestMapping(Constant.API_VERSION + "graphspaces/{graphspace}/graphs/{graph}/text2gql") +public class Text2GqlController extends BaseController { + + @Autowired + private Text2GqlService service; + + @Autowired + private Text2GqlSettingsService settings; + + @Autowired + private MessageSourceHandler messages; + + @PostMapping + public Text2GqlResult generate(@PathVariable("graphspace") String graphSpace, + @PathVariable("graph") String graph, + @RequestBody Text2GqlRequest request) { + HugeClient client = this.authClient(graphSpace, graph); + return this.service.generate(client, this.resolveRequest(request)); + } + + @PostMapping(value = "stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter stream(@PathVariable("graphspace") String graphSpace, + @PathVariable("graph") String graph, + @RequestBody Text2GqlRequest request, + HttpServletResponse response) { + HugeClient client = this.authClient(graphSpace, graph); + HttpServletRequest servletRequest = this.getRequest(); + Text2GqlService.PreparedGeneration prepared; + try { + prepared = this.service.prepare(client, this.resolveRequest(request)); + } finally { + // ResponseBodyAdvice does not own SseEmitter completion. No graph client escapes to callbacks. + servletRequest.removeAttribute("hugeClient"); + client.close(); + } + Map<String, String> safeMessages = new LinkedHashMap<>(); + for (String key : new String[]{"text2gql.model-failed", "text2gql.model-timeout", + "text2gql.model-authentication-failed", "text2gql.model-rate-limited"}) { + safeMessages.put(key, this.messages.getMessage(key)); + } + response.setHeader("Cache-Control", "no-cache"); + response.setHeader("X-Accel-Buffering", "no"); + SseEmitter emitter = new SseEmitter(this.service.streamTimeoutMillis()); + StreamEvents events = new StreamEvents(emitter, Map.copyOf(safeMessages)); + emitter.onTimeout(events::timeout); + emitter.onError(failure -> events.cancel()); + emitter.onCompletion(events::cancel); + events.waiting(); Review Comment: Minor: After `waiting`, this emitter writes only on content deltas (`returnThinking(false)` drops reasoning chunks), and the container reports a closed client only when a write fails. On embedded Tomcat, after the client closed the socket, none of `onCompletion`, `onError` or `onTimeout` fired within 10 s; the next `send` threw `AsyncRequestNotUsableException`. So a cancel while a reasoning model thinks, or before a slow first token, leaves the OkHttp call running until the first delta or the 60 s deadline, on the shared key when the shared model is used. It is bounded, so not blocking. An SSE comment heartbeat every few seconds while nothing has been written would turn the disconnect into a failed write and cancel the call. A fixture that sleeps before its first chunk can check that its connection closes well before the deadline. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
