# edited by glg import time from pypos.modules.sinkronisasi.services.export_cycle_service import ExportCycleService from pypos.modules.sinkronisasi.services.network_orchestrator_service import ( NetworkOrchestratorService, ) class DashboardExportBatchMixin: # edited by glg # Dekomposisi hotspot controller: # orkestrasi export batch dipisah ke mixin fokus tanggung jawab export-runtime. def _export_json_batch( self, source="timer", upload_limit_override=None, requeue_failed_transient=False, ): started_at = time.perf_counter() runtime_result = {"ran": False, "skipped": "unknown"} try: if self.is_any_sync_running(): with self._export_dispatch_lock: self._pending_export_after_sync = True self.log_info("[EXPORT] Skip: sinkronisasi sedang berjalan (queued)") self._schedule_export_recheck(3000) runtime_result = {"ran": False, "skipped": "sync_running"} return runtime_result if ExportCycleService.is_running(): self.log_info("[EXPORT] Skip: export batch masih berjalan (singleton guard)") runtime_result = {"ran": False, "skipped": "already_running"} return runtime_result gate = self._can_run_network_operation( NetworkOrchestratorService.PROCESS_EXPORT, force_probe=False, log_block=False, ) skip_upload_by_orchestrator = not bool(gate.get("allow")) if skip_upload_by_orchestrator: self.log_info( "[EXPORT] Upload ditunda oleh orchestrator: " f"state={gate.get('state')} reason={gate.get('reason')}" ) rollout_guard = self._evaluate_export_rollout_runtime_guard(source=str(source or "timer")) if not bool(rollout_guard.get("allow")): self.log_warning( "[EXPORT][ROLLOUT_GUARD] BLOCK " f"reason={rollout_guard.get('reason')} " f"decision={rollout_guard.get('decision')} " f"healthy={int(bool(rollout_guard.get('healthy')))} " f"in_scope={int(bool(rollout_guard.get('in_scope')))} " f"branch_id={int(rollout_guard.get('branch_id') or 0)}" ) self._schedule_export_recheck(5000) runtime_result = { "ran": False, "skipped": "rollout_guard_halt", "rollout_guard": rollout_guard, } return runtime_result cycle_service = getattr(self.view, "export_cycle_service", None) or ExportCycleService() def _run_direct_cycle( *, cycle_source, cycle_skip_upload, cycle_upload_limit_override, cycle_requeue_failed_transient, cycle_upload_timeout_override, ): return cycle_service.run_cycle( source=str(cycle_source or source or "timer"), skip_upload=bool(cycle_skip_upload), upload_limit=cycle_upload_limit_override, requeue_failed_transient=bool(cycle_requeue_failed_transient), upload_timeout_override=cycle_upload_timeout_override, ) result = {} event_runtime_service = self._get_event_first_export_runtime_service() if event_runtime_service is not None and self._is_event_ingestion_runtime_enabled(): runtime_policy = self._get_event_runtime_numeric_policy() event_runtime_metrics = self._build_event_runtime_metrics_payload(source=str(source or "timer")) machine_id, _ = self._resolve_sync_device_context() def _cycle_handler(**kwargs): return _run_direct_cycle( cycle_source=kwargs.get("source", source), cycle_skip_upload=kwargs.get("skip_upload", skip_upload_by_orchestrator), cycle_upload_limit_override=kwargs.get("upload_limit_override", upload_limit_override), cycle_requeue_failed_transient=kwargs.get( "requeue_failed_transient", bool(requeue_failed_transient), ), cycle_upload_timeout_override=kwargs.get("upload_timeout_override"), ) event_runtime_result = event_runtime_service.run_trigger_cycle( source=str(source or "timer"), source_id=str(machine_id or ""), cycle_handler=_cycle_handler, metrics=event_runtime_metrics, skip_upload=bool(skip_upload_by_orchestrator), upload_limit_override=upload_limit_override, requeue_failed_transient=bool(requeue_failed_transient), upload_timeout_override=None, dedup_window_seconds=runtime_policy.get("dedup_window_seconds"), claim_limit=runtime_policy.get("claim_limit"), lease_seconds=runtime_policy.get("lease_seconds"), max_inflight=runtime_policy.get("max_inflight"), retry_delay_seconds=runtime_policy.get("retry_delay_seconds"), purge_sent_days=runtime_policy.get("purge_sent_days"), purge_limit=runtime_policy.get("purge_limit"), release_expired_each_cycle=bool(runtime_policy.get("release_expired_each_cycle")), ) result = dict(event_runtime_result.get("cycle_result") or {}) result["event_runtime"] = { "enqueue": event_runtime_result.get("enqueue") if isinstance(event_runtime_result, dict) else {}, "drain": event_runtime_result.get("drain") if isinstance(event_runtime_result, dict) else {}, } drain_payload = result.get("event_runtime", {}).get("drain", {}) enqueue_payload = result.get("event_runtime", {}).get("enqueue", {}) if ( int(drain_payload.get("claimed") or 0) > 0 or int(drain_payload.get("failed") or 0) > 0 or int(drain_payload.get("purged_sent") or 0) > 0 or not bool(enqueue_payload.get("accepted", True)) ): self.log_info( "[EVENT_FIRST] " f"accepted={int(bool(enqueue_payload.get('accepted')))} " f"ingested={int(enqueue_payload.get('ingested_count') or 0)} " f"duplicate={int(enqueue_payload.get('duplicate_count') or 0)} " f"claimed={int(drain_payload.get('claimed') or 0)} " f"processed={int(drain_payload.get('processed') or 0)} " f"acked={int(drain_payload.get('acked') or 0)} " f"failed={int(drain_payload.get('failed') or 0)} " f"purged_sent={int(drain_payload.get('purged_sent') or 0)}" ) else: result = _run_direct_cycle( cycle_source=str(source or "timer"), cycle_skip_upload=bool(skip_upload_by_orchestrator), cycle_upload_limit_override=upload_limit_override, cycle_requeue_failed_transient=bool(requeue_failed_transient), cycle_upload_timeout_override=None, ) if not result.get("ran"): skip_reason = str(result.get("skipped") or "").strip() or "unknown" self.log_info( "[EXPORT] Skip: " f"reason={skip_reason} " f"guard_source={result.get('guard_source') or 'unknown'}" ) runtime_result = dict(result) return result exported_rows = int(result.get("exported_rows") or 0) if exported_rows: self.log_info( f"[EXPORT] Batch JSON berhasil ({str(source or 'timer')}): {exported_rows} transaksi" ) uploaded = int(result.get("uploaded") or 0) failed = int(result.get("failed") or 0) retried = int(result.get("retried") or 0) cleaned_files = int(result.get("cleaned_files") or 0) cleaned_empty_flux = int(result.get("cleaned_empty_flux") or 0) requeued_failed_transient = int(result.get("requeued_failed_transient") or 0) suppressed_direct_only = int(result.get("suppressed_direct_only") or 0) suppressed_table_toggle = int(result.get("suppressed_table_toggle") or 0) if ( uploaded or failed or retried or cleaned_files or cleaned_empty_flux or requeued_failed_transient or suppressed_direct_only or suppressed_table_toggle ): self.log_info( "[EXPORT] Upload file: " f"uploaded={uploaded} retried={retried} failed={failed} " f"cleaned_file={cleaned_files} cleaned_empty_flux={cleaned_empty_flux} " f"requeued_failed_transient={requeued_failed_transient} " f"suppressed_direct_only={suppressed_direct_only} " f"suppressed_table_toggle={suppressed_table_toggle}" ) retry_items = result.get("retry_items") if isinstance(result.get("retry_items"), list) else [] failed_items = result.get("failed_items") if isinstance(result.get("failed_items"), list) else [] for idx, item in enumerate(retry_items[:5], start=1): if not isinstance(item, dict): continue self.log_warning( "[EXPORT][OBS][RETRY] " f"#{idx} row_id={item.get('row_id')} table={item.get('table_name')} " f"phase={item.get('phase')} reason={item.get('error')} " f"idem={item.get('idempotency_key')} status_code={item.get('status_code')}" ) for idx, item in enumerate(failed_items[:5], start=1): if not isinstance(item, dict): continue self.log_warning( "[EXPORT][OBS][FAILED] " f"#{idx} row_id={item.get('row_id')} table={item.get('table_name')} " f"phase={item.get('phase')} reason={item.get('error')} " f"idem={item.get('idempotency_key')} status_code={item.get('status_code')}" ) runtime_result = dict(result) return result except ( TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError, ) as e: self.log_warning(f"[EXPORT] Batch JSON gagal: {e}") runtime_result = {"ran": False, "skipped": "exception", "error": str(e)} return runtime_result finally: elapsed_ms = (time.perf_counter() - started_at) * 1000.0 try: runtime_service = getattr(self, "_export_runtime_metrics", None) if runtime_service is not None and hasattr(runtime_service, "observe_cycle"): runtime_service.observe_cycle(runtime_result, elapsed_ms) except ( TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError, ) as exc: self.log_warning(f"[EXPORT][METRICS] gagal catat runtime: {exc}") if elapsed_ms >= 400.0: self.log_info( f"[PERF] export_cycle source={str(source or 'timer')} " f"elapsed_ms={elapsed_ms:.1f}" )