( c *gin.Context, pinned pluginruntime.PinnedEndpoint, protocolRequest pluginruntime.ProtocolRequestContext, taskID string, machine *relay.PluginResponsesMachine, deps pluginProtocolBridgeDeps, )
| 370 | } |
| 371 | |
| 372 | func streamTaskPluginProtocol( |
| 373 | c *gin.Context, |
| 374 | pinned pluginruntime.PinnedEndpoint, |
| 375 | protocolRequest pluginruntime.ProtocolRequestContext, |
| 376 | taskID string, |
| 377 | machine *relay.PluginResponsesMachine, |
| 378 | deps pluginProtocolBridgeDeps, |
| 379 | ) { |
| 380 | generation := pinned.Generation.Number |
| 381 | pluginKey := pinned.Plugin.Meta.Key |
| 382 | logger.LogDebug( |
| 383 | c, |
| 384 | "task_plugin subsystem=protocol event=observation_start generation=%d plugin=%q mode=stream public_task_id=%q timeout_ms=%d tick_ms=%d heartbeat_ms=%d", |
| 385 | generation, |
| 386 | pluginKey, |
| 387 | taskID, |
| 388 | deps.observationTimeout.Milliseconds(), |
| 389 | deps.tickInterval.Milliseconds(), |
| 390 | deps.heartbeatInterval.Milliseconds(), |
| 391 | ) |
| 392 | created, err := machine.CreatedEvent() |
| 393 | if err != nil { |
| 394 | logger.LogDebug(c, "task_plugin subsystem=protocol event=observation_failed generation=%d plugin=%q mode=stream stage=created_event reason=state_machine_error", generation, pluginKey) |
| 395 | respondPluginProtocolError(c, http.StatusInternalServerError, "task_protocol_error", "Task protocol request failed") |
| 396 | return |
| 397 | } |
| 398 | helper.SetEventStreamHeaders(c) |
| 399 | if err = writeTaskPluginProtocolEvent(c, created); err != nil { |
| 400 | logger.LogDebug(c, "task_plugin subsystem=protocol event=client_write_failed generation=%d plugin=%q mode=stream stage=created_event", generation, pluginKey) |
| 401 | return |
| 402 | } |
| 403 | |
| 404 | observationContext, cancelObservation := context.WithTimeout(c.Request.Context(), deps.observationTimeout) |
| 405 | defer cancelObservation() |
| 406 | heartbeatTicker := time.NewTicker(deps.heartbeatInterval) |
| 407 | defer heartbeatTicker.Stop() |
| 408 | |
| 409 | var previous relay.ProtocolState |
| 410 | tickNumber := uint64(0) |
| 411 | lastStatus := "" |
| 412 | for { |
| 413 | loadStarted := deps.now() |
| 414 | loadContext, cancelLoad := context.WithTimeout(observationContext, deps.loadTimeout) |
| 415 | task, exists, loadErr := deps.loadTask( |
| 416 | loadContext, |
| 417 | common.GetContextKeyInt(c, constant.ContextKeyUserId), |
| 418 | constant.TaskPlatform(pinned.Plugin.Meta.Key), |
| 419 | taskID, |
| 420 | ) |
| 421 | loadContextErr := loadContext.Err() |
| 422 | cancelLoad() |
| 423 | loadElapsed := deps.now().Sub(loadStarted) |
| 424 | if errors.Is(loadContextErr, context.DeadlineExceeded) && |
| 425 | observationContext.Err() == nil && |
| 426 | c.Request.Context().Err() == nil { |
| 427 | logger.LogWarn(c, fmt.Sprintf( |
| 428 | "task protocol database observation overloaded; plugin=%s task=%s", |
| 429 | pinned.Plugin.Meta.Key, |
no test coverage detected