From fc711517721c926bea26e1b4a3f94710ebf25993 Mon Sep 17 00:00:00 2001 From: Vitthal Mirji Date: Tue, 6 Oct 2026 18:32:07 +0200 Subject: [PATCH 1/5] feat: add closed types for error context and sql parameters --- .../flowforge/core/types/ContextValue.scala | 42 +++++++++++++++++++ .../com/flowforge/core/types/SqlValue.scala | 38 +++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 modules/core/src/main/scala/com/flowforge/core/types/ContextValue.scala create mode 100644 modules/core/src/main/scala/com/flowforge/core/types/SqlValue.scala diff --git a/modules/core/src/main/scala/com/flowforge/core/types/ContextValue.scala b/modules/core/src/main/scala/com/flowforge/core/types/ContextValue.scala new file mode 100644 index 00000000..e1ad3586 --- /dev/null +++ b/modules/core/src/main/scala/com/flowforge/core/types/ContextValue.scala @@ -0,0 +1,42 @@ +package com.flowforge.core.types + +/** + * A value attached to an error for diagnosis. + * + * Error context used to be `Map[String, Any]`, which meant anything at all could be put in it: a whole + * `DataFrame`, a credential, a mutable builder, a closure. Nothing in the type said otherwise, and nothing + * read it back except one line of string interpolation. `Any` in a public signature also costs the thing the + * rest of this codebase is built on, which is that a signature tells you what a value can be. + * + * Three cases cover what diagnostic context actually holds: a label, a count or size, and a yes/no. A caller + * that wants to attach something else has to decide how it reads first, which is the decision that was being + * skipped. + */ +sealed abstract class ContextValue extends Product with Serializable { + + /** How the value appears in an error message or a log line. */ + def render: String +} + +object ContextValue { + + /** A string: a field name, an identifier, a state name. */ + final case class Text(value: String) extends ContextValue { + def render: String = value + } + + /** + * A whole number: a row count, a byte size, an attempt number. + * + * `Long` rather than `Int` because the counts that end up here are row and byte counts, and widening is + * free at the call site while narrowing is not. + */ + final case class Num(value: Long) extends ContextValue { + def render: String = value.toString + } + + /** A yes/no: whether a retry is possible, whether a cache was warm. */ + final case class Flag(value: Boolean) extends ContextValue { + def render: String = value.toString + } +} diff --git a/modules/core/src/main/scala/com/flowforge/core/types/SqlValue.scala b/modules/core/src/main/scala/com/flowforge/core/types/SqlValue.scala new file mode 100644 index 00000000..dbbc637e --- /dev/null +++ b/modules/core/src/main/scala/com/flowforge/core/types/SqlValue.scala @@ -0,0 +1,38 @@ +package com.flowforge.core.types + +import java.time.Instant + +/** + * A value bound to a placeholder in a SQL statement. + * + * Parameters used to be `List[Any]`, which says nothing about what a driver can actually bind. The set of + * things a JDBC `PreparedStatement` accepts is small and closed, so the type can say so: a string, a whole + * number, a fractional number, a boolean, an instant, or SQL `NULL`. A caller with something else has to + * decide how it maps to a column before the call rather than after it, which is the decision `Any` let + * everyone skip. + * + * `Null` is a case rather than an absent element because a bound `NULL` still occupies a placeholder; + * dropping it from the list would shift every parameter after it. + */ +sealed abstract class SqlValue extends Product with Serializable + +object SqlValue { + + /** A character value: `VARCHAR`, `TEXT`, `CHAR`. */ + final case class Text(value: String) extends SqlValue + + /** A whole number: `INT`, `BIGINT`. `Long` so the widest integral column fits. */ + final case class Num(value: Long) extends SqlValue + + /** A fractional number: `DOUBLE`, `FLOAT`, `NUMERIC`. */ + final case class Decimal(value: Double) extends SqlValue + + /** A boolean value: `BOOLEAN`, or the single-bit column a driver maps it to. */ + final case class Flag(value: Boolean) extends SqlValue + + /** A point in time: `TIMESTAMP`, `TIMESTAMPTZ`. */ + final case class Timestamp(value: Instant) extends SqlValue + + /** SQL `NULL`, whatever the column's type. */ + case object Null extends SqlValue +} From 1a4a7b8380645151d056de5494e8aa631479ca7a Mon Sep 17 00:00:00 2001 From: Vitthal Mirji Date: Tue, 6 Oct 2026 18:32:19 +0200 Subject: [PATCH 2/5] refactor: type error context as ContextValue instead of Any --- .../safety/ConnectorErrorMapper.scala | 16 +++-- .../core/algebra/EnterpriseTableAlgebra.scala | 30 +++++---- .../flowforge/core/algebra/TypeClasses.scala | 33 +++++----- .../com/flowforge/core/types/ErrorTypes.scala | 62 +++++++++---------- .../flowforge/core/types/PipelineError.scala | 24 +++---- .../EncoderDecoderInstanceCoverageSpec.scala | 5 +- .../core/errors/ErrorTypesCoverageSpec.scala | 24 ++++--- .../FlowForgeErrorUtilsCoverageSpec.scala | 4 +- .../flowforge/core/types/ErrorTypesSpec.scala | 36 ++++++----- 9 files changed, 129 insertions(+), 105 deletions(-) diff --git a/modules/connectors/src/main/scala/com/flowforge/connectors/safety/ConnectorErrorMapper.scala b/modules/connectors/src/main/scala/com/flowforge/connectors/safety/ConnectorErrorMapper.scala index 9955139b..7d9cff48 100644 --- a/modules/connectors/src/main/scala/com/flowforge/connectors/safety/ConnectorErrorMapper.scala +++ b/modules/connectors/src/main/scala/com/flowforge/connectors/safety/ConnectorErrorMapper.scala @@ -2,7 +2,7 @@ package com.flowforge.connectors.safety import com.flowforge.core.safety.ErrorMapper import com.flowforge.core.types.FlowForgeError.{ ConfigurationError, ValidationError } -import com.flowforge.core.types.SystemError +import com.flowforge.core.types.{ ContextValue, SystemError } /** * Connector-focused ErrorMapper. @@ -13,17 +13,25 @@ import com.flowforge.core.types.SystemError object ConnectorErrorMapper { implicit val connectorMapper: ErrorMapper = { case e: java.nio.file.NoSuchFileException => - ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "NoSuchFile")) + ValidationError( + s"File not found: ${e.getMessage}", + None, + context = Map("cause" -> ContextValue.Text("NoSuchFile")), + ) .withCause(e) case e: java.io.FileNotFoundException => - ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "FileNotFound")) + ValidationError( + s"File not found: ${e.getMessage}", + None, + context = Map("cause" -> ContextValue.Text("FileNotFound")), + ) .withCause(e) case e: java.io.IOException => SystemError.ServiceUnavailable(serviceName = "filesystem", message = e.getMessage, cause = Some(e)) case e: java.sql.SQLException => ConfigurationError( s"JDBC error: ${e.getMessage}", - context = Map("sqlState" -> String.valueOf(e.getSQLState)), + context = Map("sqlState" -> ContextValue.Text(String.valueOf(e.getSQLState))), ).withCause(e) case other => ErrorMapper.default(other) } diff --git a/modules/core/src/main/scala/com/flowforge/core/algebra/EnterpriseTableAlgebra.scala b/modules/core/src/main/scala/com/flowforge/core/algebra/EnterpriseTableAlgebra.scala index 294bd4cb..d39f4b5f 100644 --- a/modules/core/src/main/scala/com/flowforge/core/algebra/EnterpriseTableAlgebra.scala +++ b/modules/core/src/main/scala/com/flowforge/core/algebra/EnterpriseTableAlgebra.scala @@ -40,7 +40,7 @@ import cats.data.{ NonEmptyList, ValidatedNel } import cats.effect.{ MonadCancel, Resource } import cats.implicits._ import com.flowforge.core.types.RefinedTypes.{ FieldName, TableName } -import com.flowforge.core.types.{ ErrorCategory, ErrorSeverity, FlowForgeError } +import com.flowforge.core.types.{ ContextValue, ErrorCategory, ErrorSeverity, FlowForgeError } import java.time.Instant import scala.concurrent.duration.FiniteDuration @@ -486,15 +486,15 @@ case class TableNotFound(tableName: TableName) extends TableError { val message = s"Table '${tableName.value}' not found" val category = ErrorCategory.System val severity = ErrorSeverity.Error - val context = Map("tableName" -> tableName.value) + val context = Map("tableName" -> ContextValue.Text(tableName.value)) val cause = None val timestamp = java.time.Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Check table name spelling", "Verify table exists", "Check permissions") - def withContext(additionalContext: Map[String, Any]) = this - def withCause(underlyingCause: Throwable) = this + def withContext(additionalContext: Map[String, ContextValue]) = this + def withCause(underlyingCause: Throwable) = this } sealed trait PartitionError extends FlowForgeError @@ -502,15 +502,15 @@ case class PartitionNotFound(partitionSpec: PartitionSpec) extends PartitionErro val message = s"Partition '${partitionSpec.toPartitionPath}' not found" val category = ErrorCategory.System val severity = ErrorSeverity.Error - val context = Map("partitionSpec" -> partitionSpec.toPartitionPath) + val context = Map("partitionSpec" -> ContextValue.Text(partitionSpec.toPartitionPath)) val cause = None val timestamp = java.time.Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Check partition specification", "Verify partition exists") - def withContext(additionalContext: Map[String, Any]) = this - def withCause(underlyingCause: Throwable) = this + def withContext(additionalContext: Map[String, ContextValue]) = this + def withCause(underlyingCause: Throwable) = this } sealed trait BlobError extends FlowForgeError @@ -519,18 +519,22 @@ case class BlobAccessError( blobName: String, reason: String) extends BlobError { - val message = s"Cannot access blob '$blobName' in bucket '$bucketName': $reason" - val category = ErrorCategory.System - val severity = ErrorSeverity.Error - val context = Map("bucketName" -> bucketName, "blobName" -> blobName, "reason" -> reason) + val message = s"Cannot access blob '$blobName' in bucket '$bucketName': $reason" + val category = ErrorCategory.System + val severity = ErrorSeverity.Error + val context = Map( + "bucketName" -> ContextValue.Text(bucketName), + "blobName" -> ContextValue.Text(blobName), + "reason" -> ContextValue.Text(reason), + ) val cause = None val timestamp = java.time.Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = true val recoveryHints = List("Check cloud permissions", "Verify bucket exists", "Retry operation") - def withContext(additionalContext: Map[String, Any]) = this - def withCause(underlyingCause: Throwable) = this + def withContext(additionalContext: Map[String, ContextValue]) = this + def withCause(underlyingCause: Throwable) = this } // =============================== diff --git a/modules/core/src/main/scala/com/flowforge/core/algebra/TypeClasses.scala b/modules/core/src/main/scala/com/flowforge/core/algebra/TypeClasses.scala index 7d23a6e5..2a16733f 100644 --- a/modules/core/src/main/scala/com/flowforge/core/algebra/TypeClasses.scala +++ b/modules/core/src/main/scala/com/flowforge/core/algebra/TypeClasses.scala @@ -753,17 +753,17 @@ object EncodingHints { */ sealed trait EncodingError extends FlowForgeError case class UnsupportedFormat(format: DataFormat, dataType: String) extends EncodingError { - val message = s"Format $format is not supported for data type $dataType" - val category = ErrorCategory.Validation - val severity = ErrorSeverity.Error - val context = Map("format" -> format.toString, "dataType" -> dataType) - val cause = None + val message = s"Format $format is not supported for data type $dataType" + val category = ErrorCategory.Validation + val severity = ErrorSeverity.Error + val context = Map("format" -> ContextValue.Text(format.toString), "dataType" -> ContextValue.Text(dataType)) + val cause = None val timestamp = Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Use a supported format", "Implement custom encoder") - def withContext(additionalContext: Map[String, Any]): EncodingError = + def withContext(additionalContext: Map[String, ContextValue]): EncodingError = copy() // Simplified for brevity def withCause(underlyingCause: Throwable): EncodingError = copy() // Simplified for brevity @@ -777,14 +777,14 @@ case class CorruptedData(details: String) extends DecodingError { val message = s"Data is corrupted: $details" val category = ErrorCategory.Validation val severity = ErrorSeverity.Error - val context = Map("details" -> details) + val context = Map("details" -> ContextValue.Text(details)) val cause = None val timestamp = Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Check data source", "Re-download data", "Use backup data") - def withContext(additionalContext: Map[String, Any]): DecodingError = + def withContext(additionalContext: Map[String, ContextValue]): DecodingError = copy() // Simplified for brevity def withCause(underlyingCause: Throwable): DecodingError = copy() // Simplified for brevity @@ -797,16 +797,17 @@ sealed trait SchemaError extends FlowForgeError case class SchemaIncompatible(expected: DataSchema, actual: DataSchema) extends SchemaError { val message = s"Schema incompatible: expected ${expected.fields.length} fields, got ${actual.fields.length}" - val category = ErrorCategory.Validation - val severity = ErrorSeverity.Error - val context = Map("expected" -> expected.toString, "actual" -> actual.toString) + val category = ErrorCategory.Validation + val severity = ErrorSeverity.Error + val context = + Map("expected" -> ContextValue.Text(expected.toString), "actual" -> ContextValue.Text(actual.toString)) val cause = None val timestamp = Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Update schema", "Enable schema evolution", "Transform data") - def withContext(additionalContext: Map[String, Any]): SchemaError = + def withContext(additionalContext: Map[String, ContextValue]): SchemaError = copy() // Simplified for brevity def withCause(underlyingCause: Throwable): SchemaError = copy() // Simplified for brevity @@ -820,14 +821,14 @@ case class SerializationFailed(reason: String) extends SerializationError { val message = s"Serialization failed: $reason" val category = ErrorCategory.System val severity = ErrorSeverity.Error - val context = Map("reason" -> reason) + val context = Map("reason" -> ContextValue.Text(reason)) val cause = None val timestamp = Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = true val recoveryHints = List("Retry operation", "Check data format", "Use alternative serializer") - def withContext(additionalContext: Map[String, Any]): SerializationError = + def withContext(additionalContext: Map[String, ContextValue]): SerializationError = copy() // Simplified for brevity def withCause(underlyingCause: Throwable): SerializationError = copy() // Simplified for brevity @@ -841,14 +842,14 @@ case class RuleViolation(ruleName: String, details: String) extends ContractViol val message = s"Contract rule '$ruleName' violated: $details" val category = ErrorCategory.Business val severity = ErrorSeverity.Error - val context = Map("rule" -> ruleName, "details" -> details) + val context = Map("rule" -> ContextValue.Text(ruleName), "details" -> ContextValue.Text(details)) val cause = None val timestamp = Instant.now() val errorId = java.util.UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Fix data quality", "Update contract rules", "Contact data owner") - def withContext(additionalContext: Map[String, Any]): ContractViolation = + def withContext(additionalContext: Map[String, ContextValue]): ContractViolation = copy() // Simplified for brevity def withCause(underlyingCause: Throwable): ContractViolation = copy() // Simplified for brevity diff --git a/modules/core/src/main/scala/com/flowforge/core/types/ErrorTypes.scala b/modules/core/src/main/scala/com/flowforge/core/types/ErrorTypes.scala index c79e48b9..bad7bb96 100644 --- a/modules/core/src/main/scala/com/flowforge/core/types/ErrorTypes.scala +++ b/modules/core/src/main/scala/com/flowforge/core/types/ErrorTypes.scala @@ -88,7 +88,7 @@ trait FlowForgeError extends RuntimeException with Product with Serializable { def severity: ErrorSeverity /** Optional error context for debugging */ - def context: Map[String, Any] + def context: Map[String, ContextValue] /** Optional underlying cause */ def cause: Option[Throwable] @@ -113,12 +113,12 @@ trait FlowForgeError extends RuntimeException with Product with Serializable { /** * Enrich this error with additional context. Returns a new error instance with merged context. */ - def withContext(additionalContext: Map[String, Any]): FlowForgeError + def withContext(additionalContext: Map[String, ContextValue]): FlowForgeError /** * Enrich this error with a single context value. */ - def withContext(key: String, value: Any): FlowForgeError = + def withContext(key: String, value: ContextValue): FlowForgeError = withContext(Map(key -> value)) /** @@ -234,7 +234,7 @@ object ValidationError { schemaName: Option[String] = None, message: String = "Schema validation failed", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -246,7 +246,7 @@ object ValidationError { "Review schema evolution settings", ) - def withContext(additionalContext: Map[String, Any]): SchemaViolation = + def withContext(additionalContext: Map[String, ContextValue]): SchemaViolation = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): SchemaViolation = @@ -262,7 +262,7 @@ object ValidationError { threshold: Option[String] = None, message: String = "Data quality constraint violated", severity: ErrorSeverity = ErrorSeverity.Warning, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -274,7 +274,7 @@ object ValidationError { "Consider data cleansing steps", ) - def withContext(additionalContext: Map[String, Any]): QualityViolation = + def withContext(additionalContext: Map[String, ContextValue]): QualityViolation = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): QualityViolation = @@ -289,7 +289,7 @@ object ValidationError { recordId: Option[String] = None, message: String = "Required field is missing", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -301,7 +301,7 @@ object ValidationError { "Review field mapping configuration", ) - def withContext(additionalContext: Map[String, Any]): MissingRequiredField = + def withContext(additionalContext: Map[String, ContextValue]): MissingRequiredField = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): MissingRequiredField = @@ -318,7 +318,7 @@ object ValidationError { value: String, message: String = "Data type mismatch", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -330,7 +330,7 @@ object ValidationError { "Update type conversion rules", ) - def withContext(additionalContext: Map[String, Any]): TypeMismatch = + def withContext(additionalContext: Map[String, ContextValue]): TypeMismatch = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): TypeMismatch = @@ -361,7 +361,7 @@ object SystemError { current: String, message: String = "System resource exhausted", severity: ErrorSeverity = ErrorSeverity.Critical, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString, @@ -375,7 +375,7 @@ object SystemError { "Retry after delay", ) - def withContext(additionalContext: Map[String, Any]): ResourceExhausted = + def withContext(additionalContext: Map[String, ContextValue]): ResourceExhausted = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): ResourceExhausted = @@ -390,7 +390,7 @@ object SystemError { endpoint: Option[String] = None, message: String = "Required service is unavailable", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString, @@ -404,7 +404,7 @@ object SystemError { "Fall back to alternative service", ) - def withContext(additionalContext: Map[String, Any]): ServiceUnavailable = + def withContext(additionalContext: Map[String, ContextValue]): ServiceUnavailable = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): ServiceUnavailable = @@ -420,7 +420,7 @@ object SystemError { elapsed: FiniteDuration, message: String = "Operation timed out", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString, @@ -434,7 +434,7 @@ object SystemError { "Retry with circuit breaker", ) - def withContext(additionalContext: Map[String, Any]): OperationTimeout = + def withContext(additionalContext: Map[String, ContextValue]): OperationTimeout = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): OperationTimeout = @@ -467,7 +467,7 @@ object BusinessError { datasetId: String, message: String = "Data contract violated", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -480,7 +480,7 @@ object BusinessError { "Implement data transformation", ) - def withContext(additionalContext: Map[String, Any]): DataContractViolation = + def withContext(additionalContext: Map[String, ContextValue]): DataContractViolation = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): DataContractViolation = @@ -497,7 +497,7 @@ object BusinessError { violationDuration: Duration, message: String = "SLA violation detected", severity: ErrorSeverity = ErrorSeverity.Warning, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -510,7 +510,7 @@ object BusinessError { "Update SLA agreements", ) - def withContext(additionalContext: Map[String, Any]): SlaViolation = + def withContext(additionalContext: Map[String, ContextValue]): SlaViolation = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): SlaViolation = @@ -533,7 +533,7 @@ object DataProcessingError { reason: String, message: String = "Data processing failed", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -546,7 +546,7 @@ object DataProcessingError { "Implement error handling in pipeline", ) - def withContext(additionalContext: Map[String, Any]): ProcessingFailure = + def withContext(additionalContext: Map[String, ContextValue]): ProcessingFailure = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): ProcessingFailure = @@ -570,7 +570,7 @@ object FlowForgeError { message: String, field: Option[String] = None, severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -585,7 +585,7 @@ object FlowForgeError { "Review validation rules", ) - def withContext(additionalContext: Map[String, Any]): ValidationError = + def withContext(additionalContext: Map[String, ContextValue]): ValidationError = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): ValidationError = @@ -599,7 +599,7 @@ object FlowForgeError { message: String, configKey: Option[String] = None, severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -614,7 +614,7 @@ object FlowForgeError { "Review configuration documentation", ) - def withContext(additionalContext: Map[String, Any]): ConfigurationError = + def withContext(additionalContext: Map[String, ContextValue]): ConfigurationError = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): ConfigurationError = @@ -629,7 +629,7 @@ object FlowForgeError { errors: NonEmptyList[FlowForgeError], message: String = "Multiple errors occurred", severity: ErrorSeverity = ErrorSeverity.Error, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, cause: Option[Throwable] = None, timestamp: Instant = Instant.now(), errorId: String = UUID.randomUUID().toString) @@ -644,7 +644,7 @@ object FlowForgeError { "Review batch processing configuration", ) - def withContext(additionalContext: Map[String, Any]): CompositeError = + def withContext(additionalContext: Map[String, ContextValue]): CompositeError = copy(context = context ++ additionalContext) def withCause(underlyingCause: Throwable): CompositeError = @@ -656,7 +656,7 @@ object FlowForgeError { */ def fromThrowable( throwable: Throwable, - context: Map[String, Any] = Map.empty, + context: Map[String, ContextValue] = Map.empty, ): SystemError.ServiceUnavailable = SystemError.ServiceUnavailable( serviceName = "unknown-service", @@ -688,7 +688,7 @@ object FlowForgeError { */ implicit val showFlowForgeError: Show[FlowForgeError] = Show.show { error => val contextStr = if (error.context.nonEmpty) { - error.context.map { case (k, v) => s"$k=$v" }.mkString(", ") + error.context.map { case (k, v) => s"$k=${v.render}" }.mkString(", ") } else { "none" } diff --git a/modules/core/src/main/scala/com/flowforge/core/types/PipelineError.scala b/modules/core/src/main/scala/com/flowforge/core/types/PipelineError.scala index a8f962d5..fc6c77c7 100644 --- a/modules/core/src/main/scala/com/flowforge/core/types/PipelineError.scala +++ b/modules/core/src/main/scala/com/flowforge/core/types/PipelineError.scala @@ -14,44 +14,44 @@ object PipelineError { val message = s"Pipeline '$name' has no stages" val category: ErrorCategory = ErrorCategory.Configuration val severity: ErrorSeverity = ErrorSeverity.Error - val context = Map("pipeline" -> name) + val context = Map("pipeline" -> ContextValue.Text(name)) val cause: Option[Throwable] = None val timestamp: Instant = Instant.now() val errorId: String = UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Add at least one stage to the pipeline") - def withContext(additionalContext: Map[String, Any]): EmptyPipeline = this - def withCause(underlyingCause: Throwable): EmptyPipeline = this + def withContext(additionalContext: Map[String, ContextValue]): EmptyPipeline = this + def withCause(underlyingCause: Throwable): EmptyPipeline = this } case class InvalidConfiguration(details: String) extends PipelineError { val message = s"Invalid pipeline configuration: $details" val category: ErrorCategory = ErrorCategory.Configuration val severity: ErrorSeverity = ErrorSeverity.Error - val context = Map("details" -> details) + val context = Map("details" -> ContextValue.Text(details)) val cause: Option[Throwable] = None val timestamp: Instant = Instant.now() val errorId: String = UUID.randomUUID().toString val isRetryable = false val recoveryHints = List("Review pipeline configuration") - def withContext(additionalContext: Map[String, Any]): InvalidConfiguration = this - def withCause(underlyingCause: Throwable): InvalidConfiguration = this + def withContext(additionalContext: Map[String, ContextValue]): InvalidConfiguration = this + def withCause(underlyingCause: Throwable): InvalidConfiguration = this } case class StageExecutionError(stageName: String, reason: String) extends PipelineError { - val message = s"Stage '$stageName' failed: $reason" - val category: ErrorCategory = ErrorCategory.System - val severity: ErrorSeverity = ErrorSeverity.Error - val context = Map("stage" -> stageName, "reason" -> reason) + val message = s"Stage '$stageName' failed: $reason" + val category: ErrorCategory = ErrorCategory.System + val severity: ErrorSeverity = ErrorSeverity.Error + val context = Map("stage" -> ContextValue.Text(stageName), "reason" -> ContextValue.Text(reason)) val cause: Option[Throwable] = None val timestamp: Instant = Instant.now() val errorId: String = UUID.randomUUID().toString val isRetryable = true val recoveryHints = List("Retry the stage", "Check stage configuration") - def withContext(additionalContext: Map[String, Any]): StageExecutionError = this - def withCause(underlyingCause: Throwable): StageExecutionError = this + def withContext(additionalContext: Map[String, ContextValue]): StageExecutionError = this + def withCause(underlyingCause: Throwable): StageExecutionError = this } } diff --git a/modules/core/src/test/scala/com/flowforge/core/algebra/EncoderDecoderInstanceCoverageSpec.scala b/modules/core/src/test/scala/com/flowforge/core/algebra/EncoderDecoderInstanceCoverageSpec.scala index 77c3519d..ee53f686 100644 --- a/modules/core/src/test/scala/com/flowforge/core/algebra/EncoderDecoderInstanceCoverageSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/algebra/EncoderDecoderInstanceCoverageSpec.scala @@ -23,9 +23,10 @@ class EncoderDecoderInstanceCoverageSpec extends AnyFunSuite with Matchers { test("EncodingError/DecodingError helpers") { val ue = UnsupportedFormat(DataFormat.Parquet, "X") - .withContext(Map("k" -> 1)).withCause(new RuntimeException("x")) + .withContext(Map("k" -> ContextValue.Num(1))).withCause(new RuntimeException("x")) ue.message.toLowerCase should include("not supported") - val cd = CorruptedData("bad").withContext(Map("k" -> 2)).withCause(new RuntimeException("y")) + val cd = + CorruptedData("bad").withContext(Map("k" -> ContextValue.Num(2))).withCause(new RuntimeException("y")) cd.message.toLowerCase should include("corrupted") } } diff --git a/modules/core/src/test/scala/com/flowforge/core/errors/ErrorTypesCoverageSpec.scala b/modules/core/src/test/scala/com/flowforge/core/errors/ErrorTypesCoverageSpec.scala index fce98ff3..2e8cca75 100644 --- a/modules/core/src/test/scala/com/flowforge/core/errors/ErrorTypesCoverageSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/errors/ErrorTypesCoverageSpec.scala @@ -1,5 +1,6 @@ package com.flowforge.core.errors import com.flowforge.core.types.BusinessError._ +import com.flowforge.core.types.ContextValue import com.flowforge.core.types.SystemError._ import com.flowforge.core.types.ValidationError._ import org.scalatest.funsuite.AnyFunSuite @@ -10,12 +11,15 @@ import scala.concurrent.duration._ class ErrorTypesCoverageSpec extends AnyFunSuite with Matchers { test("ValidationError variants support helpers") { val sv = - SchemaViolation("f", "string", "int").withContext(Map("k" -> 1)).withCause(new RuntimeException("x")) + SchemaViolation("f", "string", "int") + .withContext(Map("k" -> ContextValue.Num(1))).withCause(new RuntimeException("x")) val qv = QualityViolation("rule", "v", Some("th")) - .withContext(Map("k" -> 2)).withCause(new RuntimeException("y")) - val mrf = MissingRequiredField("id").withContext(Map("k" -> 3)).withCause(new RuntimeException("z")) + .withContext(Map("k" -> ContextValue.Num(2))).withCause(new RuntimeException("y")) + val mrf = MissingRequiredField("id") + .withContext(Map("k" -> ContextValue.Num(3))).withCause(new RuntimeException("z")) val tm = - TypeMismatch("f", "int", "string", "v").withContext(Map("k" -> 4)).withCause(new RuntimeException("w")) + TypeMismatch("f", "int", "string", "v") + .withContext(Map("k" -> ContextValue.Num(4))).withCause(new RuntimeException("w")) sv.message should include("Schema") qv.recoveryHints.nonEmpty shouldBe true mrf.fieldName shouldBe "id" @@ -24,11 +28,12 @@ class ErrorTypesCoverageSpec extends AnyFunSuite with Matchers { test("SystemError variants support helpers") { val re = - ResourceExhausted("mem", "1g", "2g").withContext(Map("h" -> true)).withCause(new RuntimeException("x")) + ResourceExhausted("mem", "1g", "2g") + .withContext(Map("h" -> ContextValue.Flag(true))).withCause(new RuntimeException("x")) val su = ServiceUnavailable("svc", Some("http://")) - .withContext(Map("h" -> false)).withCause(new RuntimeException("y")) + .withContext(Map("h" -> ContextValue.Flag(false))).withCause(new RuntimeException("y")) val ot = OperationTimeout("op", 1.second, 2.seconds) - .withContext(Map("t" -> 2)).withCause(new RuntimeException("z")) + .withContext(Map("t" -> ContextValue.Num(2))).withCause(new RuntimeException("z")) re.isRetryable shouldBe true su.serviceName shouldBe "svc" ot.timeout shouldBe 1.second @@ -36,9 +41,10 @@ class ErrorTypesCoverageSpec extends AnyFunSuite with Matchers { test("BusinessError variants support helpers") { val dc = - DataContractViolation("c", "r", "ds").withContext(Map("a" -> 1)).withCause(new RuntimeException("x")) + DataContractViolation("c", "r", "ds") + .withContext(Map("a" -> ContextValue.Num(1))).withCause(new RuntimeException("x")) val sla = SlaViolation("latency", "10ms", "50ms", java.time.Duration.ofSeconds(5)) - .withContext(Map("b" -> 2)).withCause(new RuntimeException("y")) + .withContext(Map("b" -> ContextValue.Num(2))).withCause(new RuntimeException("y")) dc.message should include("contract") sla.violationDuration shouldBe java.time.Duration.ofSeconds(5) } diff --git a/modules/core/src/test/scala/com/flowforge/core/errors/FlowForgeErrorUtilsCoverageSpec.scala b/modules/core/src/test/scala/com/flowforge/core/errors/FlowForgeErrorUtilsCoverageSpec.scala index 60d02476..533d0a9a 100644 --- a/modules/core/src/test/scala/com/flowforge/core/errors/FlowForgeErrorUtilsCoverageSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/errors/FlowForgeErrorUtilsCoverageSpec.scala @@ -1,13 +1,13 @@ package com.flowforge.core.errors -import com.flowforge.core.types.{ FlowForgeError, ValidationError } +import com.flowforge.core.types.{ ContextValue, FlowForgeError, ValidationError } import org.scalatest.funsuite.AnyFunSuite import org.scalatest.matchers.should.Matchers class FlowForgeErrorUtilsCoverageSpec extends AnyFunSuite with Matchers { test("fromThrowable and factory helpers") { val ex = new RuntimeException("boom") - val err = FlowForgeError.fromThrowable(ex, Map("k" -> "v")) + val err = FlowForgeError.fromThrowable(ex, Map("k" -> ContextValue.Text("v"))) err.message should include("boom") val mf = FlowForgeError.missingField("id") diff --git a/modules/core/src/test/scala/com/flowforge/core/types/ErrorTypesSpec.scala b/modules/core/src/test/scala/com/flowforge/core/types/ErrorTypesSpec.scala index 899e8cce..04eae1cf 100644 --- a/modules/core/src/test/scala/com/flowforge/core/types/ErrorTypesSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/types/ErrorTypesSpec.scala @@ -93,8 +93,8 @@ class ErrorTypesSpec extends AnyFunSuite with Matchers { test("ValidationError.SchemaViolation should support withContext") { val error = ValidationError.SchemaViolation("userId", "String", "Int") - val enriched = error.withContext("pipelineId", "test-pipeline") - enriched.context should contain("pipelineId" -> "test-pipeline") + val enriched = error.withContext("pipelineId", ContextValue.Text("test-pipeline")) + enriched.context should contain("pipelineId" -> ContextValue.Text("test-pipeline")) } test("ValidationError.SchemaViolation should support withCause") { @@ -280,8 +280,9 @@ class ErrorTypesSpec extends AnyFunSuite with Matchers { test("FlowForgeError.fromThrowable should include context") { val throwable = new RuntimeException("Error") - val context = Map("pipelineId" -> "test-123", "stage" -> "transform") - val error = FlowForgeError.fromThrowable(throwable, context) + val context = + Map("pipelineId" -> ContextValue.Text("test-123"), "stage" -> ContextValue.Text("transform")) + val error = FlowForgeError.fromThrowable(throwable, context) error.context shouldBe context } @@ -320,23 +321,25 @@ class ErrorTypesSpec extends AnyFunSuite with Matchers { } test("All errors should support withContext with map") { - val error = FlowForgeError.ValidationError("Error") - val enriched = error.withContext(Map("key1" -> "value1", "key2" -> "value2")) - enriched.context should contain("key1" -> "value1") - enriched.context should contain("key2" -> "value2") + val error = FlowForgeError.ValidationError("Error") + val enriched = + error.withContext(Map("key1" -> ContextValue.Text("value1"), "key2" -> ContextValue.Text("value2"))) + enriched.context should contain("key1" -> ContextValue.Text("value1")) + enriched.context should contain("key2" -> ContextValue.Text("value2")) } test("All errors should support withContext with key-value") { val error = FlowForgeError.ValidationError("Error") - val enriched = error.withContext("pipelineId", "test-pipeline") - enriched.context should contain("pipelineId" -> "test-pipeline") + val enriched = error.withContext("pipelineId", ContextValue.Text("test-pipeline")) + enriched.context should contain("pipelineId" -> ContextValue.Text("test-pipeline")) } test("Errors should preserve existing context when adding new context") { - val error = FlowForgeError.ValidationError("Error", context = Map("existing" -> "value")) - val enriched = error.withContext("new", "data") - enriched.context should contain("existing" -> "value") - enriched.context should contain("new" -> "data") + val error = + FlowForgeError.ValidationError("Error", context = Map("existing" -> ContextValue.Text("value"))) + val enriched = error.withContext("new", ContextValue.Text("data")) + enriched.context should contain("existing" -> ContextValue.Text("value")) + enriched.context should contain("new" -> ContextValue.Text("data")) } test("getMessage should return message") { @@ -362,12 +365,13 @@ class ErrorTypesSpec extends AnyFunSuite with Matchers { // Removed: FlowForgeError Show instance tests - Show instance not available // Test that FlowForgeError properties can be accessed directly test("FlowForgeError should have accessible properties") { - val error = FlowForgeError.ValidationError("Test error", context = Map("key" -> "value")) + val error = + FlowForgeError.ValidationError("Test error", context = Map("key" -> ContextValue.Text("value"))) error.message should include("Test error") error.category shouldBe ErrorCategory.Validation error.severity shouldBe ErrorSeverity.Error - error.context should contain("key" -> "value") + error.context should contain("key" -> ContextValue.Text("value")) error.isRetryable shouldBe false error.errorId should not be empty } From 91499f1d44ecac059cd06b3761f15d07f0f23a7b Mon Sep 17 00:00:00 2001 From: Vitthal Mirji Date: Tue, 6 Oct 2026 18:32:26 +0200 Subject: [PATCH 3/5] fix: remove unsound casts from core patterns --- .../core/patterns/CommonValidations.scala | 19 +++++-- .../core/patterns/DataQualityValidation.scala | 10 ++-- .../core/patterns/ReaderPattern.scala | 53 ++++++++++++------- .../core/patterns/CommonValidationsSpec.scala | 3 ++ .../patterns/DataQualityValidationSpec.scala | 10 ++-- .../core/patterns/ReaderPatternSpec.scala | 4 +- 6 files changed, 67 insertions(+), 32 deletions(-) diff --git a/modules/core/src/main/scala/com/flowforge/core/patterns/CommonValidations.scala b/modules/core/src/main/scala/com/flowforge/core/patterns/CommonValidations.scala index fc6e200c..0ea7656b 100644 --- a/modules/core/src/main/scala/com/flowforge/core/patterns/CommonValidations.scala +++ b/modules/core/src/main/scala/com/flowforge/core/patterns/CommonValidations.scala @@ -83,7 +83,6 @@ package com.flowforge.core.patterns import cats.syntax.all._ import com.flowforge.core.patterns.ValidationTypes._ -import com.flowforge.core.types.RefinedTypes._ import com.flowforge.core.types._ /** @@ -146,10 +145,22 @@ object CommonValidations { if (data.nonEmpty) { valid(data) } else { - val violation = QualityConstraint.NotNull( - FieldName.unsafeFrom("data"), + // The cast here was hiding a mismatch in the error type, not the value type: this built a + // `QualityConstraint.NotNull`, which is a constraint rather than a violation, so `invalid` produced a + // `ValidatedNel[QualityConstraint, _]` and the cast forced it into the declared result. Reporting an + // actual `QualityViolation` is what the signature promised all along. + // + // `severity` is passed because `QualityViolation` defaults it to `Warning`, while the constraint this + // replaces defaulted to `Error`. This branch rejects the dataset, so anything routing or alerting on + // severity has to see a failure rather than a warning. + invalid( + ValidationError.QualityViolation( + constraint = "nonEmpty", + violatedValue = "data", + message = "Dataset is empty", + severity = ErrorSeverity.Error, + ), ) - invalid(violation).asInstanceOf[QualityValidationResult[List[A]]] } /** diff --git a/modules/core/src/main/scala/com/flowforge/core/patterns/DataQualityValidation.scala b/modules/core/src/main/scala/com/flowforge/core/patterns/DataQualityValidation.scala index 43ab483c..9ab6b29d 100644 --- a/modules/core/src/main/scala/com/flowforge/core/patterns/DataQualityValidation.scala +++ b/modules/core/src/main/scala/com/flowforge/core/patterns/DataQualityValidation.scala @@ -16,9 +16,15 @@ object DataQualityValidation { /** * Validate data freshness - ensure data is not older than specified duration. + * + * Takes the data it is vouching for and hands it back, the same way [[completeness]] does, so the check + * composes into a validation chain. It used to claim `QualityValidationResult[A]` while having no `A` to + * return, and bridged the gap with `().asInstanceOf[A]`. That threw `ClassCastException` on the success + * path for every `A` except `Unit`, which is why every caller pinned `freshness[Unit]`. */ def freshness[A]( fieldName: String, + value: A, timestamp: Instant, maxAge: FiniteDuration, ): QualityValidationResult[A] = { @@ -27,9 +33,7 @@ object DataQualityValidation { val maxAgeInMillis = maxAge.toMillis if (age.toMillis <= maxAgeInMillis) { - // Can't return A without having an A - this method signature needs fixing - // For now, return a unit value cast to A as placeholder - ().asInstanceOf[A].validNel + value.validNel } else { val violation = QualityViolation( constraint = "freshness", diff --git a/modules/core/src/main/scala/com/flowforge/core/patterns/ReaderPattern.scala b/modules/core/src/main/scala/com/flowforge/core/patterns/ReaderPattern.scala index a22ea31f..0462f765 100644 --- a/modules/core/src/main/scala/com/flowforge/core/patterns/ReaderPattern.scala +++ b/modules/core/src/main/scala/com/flowforge/core/patterns/ReaderPattern.scala @@ -1,4 +1,3 @@ -// scalafix:off DisableSyntax.null package com.flowforge.core.patterns import cats.data.{ Kleisli, Reader, ReaderT } @@ -75,13 +74,17 @@ object ReaderPattern { def config: PipelineConfig = core.config /** - * Get environment-specific configuration + * Get environment-specific configuration. + * + * Returns a `String` because `PipelineConfig.settings` is a `Map[String, String]`. The signature used to + * be `envConfig[A]` with an `asInstanceOf[Option[A]]` behind it, so asking for anything but a `String` + * compiled and then threw `ClassCastException` at the point of use. Parsing a setting into a richer type + * belongs to the caller, which is the only place that knows what the key means. */ - def envConfig[A](key: String): Option[A] = + def envConfig(key: String): Option[String] = core.config.settings .get(s"${environment.toString.toLowerCase}.$key") .orElse(core.config.settings.get(key)) - .asInstanceOf[Option[A]] } // =============================== @@ -212,7 +215,15 @@ object ReaderPattern { body: String, ): F[Unit] def sendSlack(channel: String, message: String): F[Unit] - def sendWebhook(url: String, payload: Map[String, Any]): F[Unit] + + /** + * Post a payload to a webhook. + * + * The payload is string-valued, like every other property map in this file. It was `Map[String, Any]`, + * which meant the type promised a shape no sender could serialise without guessing. A caller that needs + * nesting should encode it and pass the encoded document under one key. + */ + def sendWebhook(url: String, payload: Map[String, String]): F[Unit] } /** Alert definitions and their lifecycle. */ @@ -254,8 +265,8 @@ object ReaderPattern { /** A single database connection: query, execute and batch. */ trait Connection[F[_]] { - def query[A](sql: String, params: List[Any]): F[List[A]] - def execute(sql: String, params: List[Any]): F[Int] + def query[A](sql: String, params: List[SqlValue]): F[List[A]] + def execute(sql: String, params: List[SqlValue]): F[Int] def batch(operations: List[SqlOperation]): F[List[Int]] } @@ -306,12 +317,12 @@ object ReaderPattern { /** * Access configuration with environment awareness */ - def getConfig[F[_]: Applicative, A]( + def getConfig[F[_]: Applicative]( key: String, - default: A, - ): FlowForgeReaderT[F, A] = + default: String, + ): FlowForgeReaderT[F, String] = ReaderT { context => - val config = context.envConfig[A](key).getOrElse(default) + val config = context.envConfig(key).getOrElse(default) Applicative[F].pure(config) } @@ -562,7 +573,7 @@ object ReaderPattern { final case class ResourceInfo( name: String, status: ResourceStatus, - config: Map[String, Any], + config: Map[String, String], lastHealthCheck: Instant) sealed trait ResourceStatus extends Product with Serializable @@ -627,7 +638,7 @@ object ReaderPattern { case class SqlOperation( sql: String, - params: List[Any]) + params: List[SqlValue]) // =============================== // TEST IMPLEMENTATIONS @@ -698,9 +709,15 @@ object ReaderPattern { def mockResourceManager[F[_]: EffectSystem]: ResourceManager[F] = new ResourceManager[F] { def acquireResource[R](name: String, config: ResourceConfig): FlowforgeResource[F, R] = { - // Safe no-op resource that yields a null placeholder of requested type R + // A mock has no way to produce an `R`, so it fails on acquire instead of pretending. It used to hand + // back `null.asInstanceOf[R]`, which left the null to surface somewhere further along, after the + // acquire had already reported success. val F = implicitly[EffectSystem[F]] - FlowforgeResource.make(F.pure(null.asInstanceOf[R]))(_ => F.unit) + FlowforgeResource.make( + F.raiseError[R]( + new UnsupportedOperationException(s"mock resource manager cannot acquire '$name'"), + ), + )(_ => F.unit) } def releaseResource(name: String): F[Unit] = implicitly[EffectSystem[F]].delay(()) def listResources: F[List[ResourceInfo]] = implicitly[EffectSystem[F]].delay(List.empty) @@ -712,8 +729,8 @@ object ReaderPattern { def withConnection[A](operation: Connection[F] => F[A]): F[A] = { val F = implicitly[EffectSystem[F]] val conn = new Connection[F] { - def query[A](sql: String, params: List[Any]): F[List[A]] = F.pure(Nil) - def execute(sql: String, params: List[Any]): F[Int] = F.pure(0) + def query[A](sql: String, params: List[SqlValue]): F[List[A]] = F.pure(Nil) + def execute(sql: String, params: List[SqlValue]): F[Int] = F.pure(0) def batch(operations: List[SqlOperation]): F[List[Int]] = F.pure(List.fill(operations.size)(0)) } @@ -783,7 +800,7 @@ object ReaderPattern { implicitly[EffectSystem[F]].delay(()) def sendSlack(channel: String, message: String): F[Unit] = implicitly[EffectSystem[F]].delay(()) - def sendWebhook(url: String, payload: Map[String, Any]): F[Unit] = + def sendWebhook(url: String, payload: Map[String, String]): F[Unit] = implicitly[EffectSystem[F]].delay(()) } diff --git a/modules/core/src/test/scala/com/flowforge/core/patterns/CommonValidationsSpec.scala b/modules/core/src/test/scala/com/flowforge/core/patterns/CommonValidationsSpec.scala index 9e177aef..a81e9e4b 100644 --- a/modules/core/src/test/scala/com/flowforge/core/patterns/CommonValidationsSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/patterns/CommonValidationsSpec.scala @@ -207,6 +207,9 @@ class CommonValidationsSpec extends AnyFunSuite with Matchers { val result = CommonValidations.validateDataQuality(data, rules) result.isInvalid shouldBe true + // The severity is asserted because `QualityViolation` defaults it to `Warning`, and a rejected dataset + // reported as a warning is the kind of thing that routes an alert to the wrong place. + result.swap.toOption.map(_.head.severity) shouldBe Some(ErrorSeverity.Error) } test("validateDataQuality works with different data types") { diff --git a/modules/core/src/test/scala/com/flowforge/core/patterns/DataQualityValidationSpec.scala b/modules/core/src/test/scala/com/flowforge/core/patterns/DataQualityValidationSpec.scala index cade01d4..69b54d3d 100644 --- a/modules/core/src/test/scala/com/flowforge/core/patterns/DataQualityValidationSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/patterns/DataQualityValidationSpec.scala @@ -21,19 +21,19 @@ class DataQualityValidationSpec extends AnyFunSuite with Matchers { test("freshness validates recent timestamp") { val recentTimestamp = Instant.now().minusSeconds(30) - val result = DataQualityValidation.freshness[Unit]("field", recentTimestamp, 1.minute) - result.isValid shouldBe true + val result = DataQualityValidation.freshness("field", "row", recentTimestamp, 1.minute) + result.toOption shouldBe Some("row") } test("freshness validates timestamp at exact max age") { val timestamp = Instant.now().minusSeconds(60) - val result = DataQualityValidation.freshness[Unit]("field", timestamp, 1.minute) + val result = DataQualityValidation.freshness("field", "row", timestamp, 1.minute) result.isValid shouldBe true } test("freshness rejects old timestamp") { val oldTimestamp = Instant.now().minusSeconds(120) - val result = DataQualityValidation.freshness[Unit]("field", oldTimestamp, 1.minute) + val result = DataQualityValidation.freshness("field", "row", oldTimestamp, 1.minute) result.isInvalid shouldBe true getErrors(result).map(_.head) match { @@ -49,7 +49,7 @@ class DataQualityValidationSpec extends AnyFunSuite with Matchers { test("freshness validates with millisecond precision") { val timestamp = Instant.now().minusMillis(500) - val result = DataQualityValidation.freshness[Unit]("field", timestamp, 1.second) + val result = DataQualityValidation.freshness("field", "row", timestamp, 1.second) result.isValid shouldBe true } diff --git a/modules/core/src/test/scala/com/flowforge/core/patterns/ReaderPatternSpec.scala b/modules/core/src/test/scala/com/flowforge/core/patterns/ReaderPatternSpec.scala index 9485fb18..d60c3fc1 100644 --- a/modules/core/src/test/scala/com/flowforge/core/patterns/ReaderPatternSpec.scala +++ b/modules/core/src/test/scala/com/flowforge/core/patterns/ReaderPatternSpec.scala @@ -193,8 +193,8 @@ class ReaderPatternSpec extends AnyFunSuite with Matchers { Map("batch.size" -> "10", "testing.batch.size" -> "99"), ) - Operations.getConfig[IO, String]("batch.size", "0").run(ctx).unsafeRunSync() shouldBe "99" - Operations.getConfig[IO, String]("missing.key", "fallback").run(ctx).unsafeRunSync() shouldBe + Operations.getConfig[IO]("batch.size", "0").run(ctx).unsafeRunSync() shouldBe "99" + Operations.getConfig[IO]("missing.key", "fallback").run(ctx).unsafeRunSync() shouldBe "fallback" } From b0940d0fae8ada5d8fb82ac4f4f588a635ff102a Mon Sep 17 00:00:00 2001 From: Vitthal Mirji Date: Tue, 6 Oct 2026 18:32:26 +0200 Subject: [PATCH 4/5] refactor: drop Any from contract and algebra signatures --- .../flowforge/contracts/DataContract.scala | 21 ++++++++++--------- .../flowforge/core/algebra/DataAlgebra.scala | 7 ++++++- .../engines/spark/SparkDataAlgebra.scala | 4 +++- 3 files changed, 20 insertions(+), 12 deletions(-) diff --git a/modules/contracts/src/main/scala/com/flowforge/contracts/DataContract.scala b/modules/contracts/src/main/scala/com/flowforge/contracts/DataContract.scala index ada3d1e2..e182bd3d 100644 --- a/modules/contracts/src/main/scala/com/flowforge/contracts/DataContract.scala +++ b/modules/contracts/src/main/scala/com/flowforge/contracts/DataContract.scala @@ -76,12 +76,11 @@ object FieldType { /** Algebraic data type for field‑level constraints. */ sealed trait FieldConstraint object FieldConstraint { - case class MinLength(value: Int) extends FieldConstraint - case class MaxLength(value: Int) extends FieldConstraint - case class Range(min: Double, max: Double) extends FieldConstraint - case class Pattern(regex: Regex) extends FieldConstraint - case class OneOf(values: Set[String]) extends FieldConstraint - case class Custom(name: String, validator: Any => Boolean) extends FieldConstraint + case class MinLength(value: Int) extends FieldConstraint + case class MaxLength(value: Int) extends FieldConstraint + case class Range(min: Double, max: Double) extends FieldConstraint + case class Pattern(regex: Regex) extends FieldConstraint + case class OneOf(values: Set[String]) extends FieldConstraint } /** Composable validation rules evaluated at runtime. */ @@ -100,19 +99,21 @@ object RuleSeverity { /** Common validation rules provided out of the box. */ object ValidationRules { - def nonNull[A](fieldName: String)(extract: A => Any): ValidationRule[A] = + def nonNull[A](fieldName: String)(extract: A => AnyRef): ValidationRule[A] = new ValidationRule[A] { val name = s"nonNull($fieldName)" def validate(data: A): ValidatedNel[ContractViolation, Unit] = - // `extract` returns Any because it reaches into caller data this module does not control, so a null - // genuinely can arrive here. Option is the narrowest way to ask without naming the literal. + // `extract` returns a reference because that is the only thing a null can be, and this rule exists to + // catch a null arriving from caller data the module does not control. It was `A => Any`, which also + // accepted a primitive the check could never fail for. `Option` is the narrowest way to ask without + // naming the literal. Option(extract(data)) match { case Some(_) => ().validNel case None => ContractViolation.NullValue(fieldName).invalidNel } } - def unique[A](fieldName: String)(extract: A => Any): ValidationRule[A] = + def unique[A, B](fieldName: String)(@annotation.unused extract: A => B): ValidationRule[A] = new ValidationRule[A] { val name = s"unique($fieldName)" def validate(data: A): ValidatedNel[ContractViolation, Unit] = diff --git a/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala b/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala index 953cd2b8..05b5f3d0 100644 --- a/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala +++ b/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala @@ -455,13 +455,18 @@ object DataAlgebra { /** * Data profile for understanding dataset characteristics. + * + * `statistics` is numeric because a profile's statistics are: min, max, mean, stddev, percentiles. It was + * `Map[String, Any]`, and every one of the three implementations passes `Map.empty`, so nothing was relying + * on the wider type. A profile that needs to report a non-numeric statistic should say what that is rather + * than reopening the map to anything. */ case class DataProfile[A]( recordCount: Long, nullCount: Long, distinctCount: Long, schema: DataSchema, - statistics: Map[String, Any]) + statistics: Map[String, Double]) /** * Schema migration for evolution. diff --git a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala index 70abaaa4..3a7eee23 100644 --- a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala +++ b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala @@ -18,7 +18,9 @@ private object ReflectionCasting { * None means the adapter returned something else. The caller treats that the same as the adapter not being * on the classpath at all, so there is nothing here worth raising. */ - def castQualityResult[A](result: Any): Option[DataAlgebra.QualityResult[DataAlgebra.Dataset[A]]] = + // `AnyRef` rather than `Any`: the argument is whatever `Method.invoke` returned, and that is an + // `Object`. Nothing wider can reach here. + def castQualityResult[A](result: AnyRef): Option[DataAlgebra.QualityResult[DataAlgebra.Dataset[A]]] = result match { case qr: DataAlgebra.QualityResult[_] => // ARCHITECTURAL: Reflection guarantees type safety here - required for modular quality system From eda8f5666972b708a03980d8cdc085939191f137 Mon Sep 17 00:00:00 2001 From: Vitthal Mirji Date: Tue, 6 Oct 2026 18:32:26 +0200 Subject: [PATCH 5/5] ci: fail the build on Any or asInstanceOf in production code --- .github/workflows/ci.yml | 5 +++ .scalafix.conf | 8 ++--- scripts/check-type-safety.sh | 69 ++++++++++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 4 deletions(-) create mode 100755 scripts/check-type-safety.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a6c342f1..1ecedd7a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,6 +61,11 @@ jobs: exit 1 fi + - name: Guard - ban Any and asInstanceOf in production sources + run: | + chmod +x scripts/check-type-safety.sh + ./scripts/check-type-safety.sh + - name: compile all run: sbt clean compileAll diff --git a/.scalafix.conf b/.scalafix.conf index cf7f93e1..6e08ffe3 100644 --- a/.scalafix.conf +++ b/.scalafix.conf @@ -13,7 +13,8 @@ DisableSyntax.noThrows = true DisableSyntax.noNulls = true DisableSyntax.noReturns = true DisableSyntax.noWhileLoops = true -# asInstanceOf checked with whitelist in CI (scalafix doesn't support per-file excludes) +# asInstanceOf and Any are banned by scripts/check-type-safety.sh, which runs in CI. They need a per-file +# allowlist for the reflection boundaries, and DisableSyntax has no per-file excludes. DisableSyntax.noIsInstanceOf = true DisableSyntax.noXml = true DisableSyntax.regex = [ @@ -67,6 +68,5 @@ OrganizeImports.groupedImports = Merge # default Ascii order the two formatters each undo the other's selector order and neither check can pass. OrganizeImports.importSelectorsOrder = SymbolsFirst -# Note: Scalafix DisableSyntax doesn't support per-file excludes -# Whitelists for asInstanceOf and scala.util.Try are enforced via shell scripts in CI -# Examples directory is excluded from CI checks via grep filter (demo code, not production) +# Note: Scalafix DisableSyntax doesn't support per-file excludes, so the checks that need an allowlist live +# in scripts/check-type-safety.sh instead. Examples are demo code and excluded from both. diff --git a/scripts/check-type-safety.sh b/scripts/check-type-safety.sh new file mode 100755 index 00000000..da1b508e --- /dev/null +++ b/scripts/check-type-safety.sh @@ -0,0 +1,69 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Bans `Any` and `asInstanceOf` from production sources. +# +# `Any` in a signature says a value can be anything, which is the one thing every other type in this +# codebase refuses to say: the caller learns nothing and the callee has to guess. `asInstanceOf` is the same +# claim made at a single expression, with a `ClassCastException` instead of a compile error when it is wrong. +# +# This is a script rather than a scalafix rule because `DisableSyntax` has no per-file excludes, and the +# allowlist is the whole reason the check is usable: a few files genuinely sit at a boundary where the JVM +# hands back an `Object`. + +# Files allowed to use either construct, each with the reason it has no narrower option. +# +# DeequAdapter reflection into Deequ, loaded by name to keep it an optional dependency. +# `Method.invoke` returns `Object`. +# SparkDataAlgebra reads a reflectively-obtained quality result back as the typed one. Same boundary. +# EffectInstances `ZIO.async[Any, Throwable, A]`: ZIO's first parameter is the environment, and `Any` +# there means the effect requires nothing. The opposite of an escape hatch. +# MemoryProfiler `Any` appears inside a string template the profiler generates sources from, and in +# a synthetic data generator. Neither is a signature anyone calls. +# OpenLineageEmitter `url.openConnection().asInstanceOf[HttpURLConnection]`: the JDK returns the +# supertype and expects the caller to narrow it. +allowlist=( + "modules/quality-deequ/src/main/scala/com/flowforge/quality/deequ/DeequAdapter.scala" + "modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala" + "modules/core/src/main/scala/com/flowforge/core/instances/EffectInstances.scala" + "modules/performance-benchmarks/src/main/scala/com/flowforge/performance/MemoryProfiler.scala" + "modules/core/src/main/scala/com/flowforge/core/lineage/OpenLineageEmitter.scala" +) + +# `examples` is demo code, excluded from the production guards the same way the println check excludes it. +find_args=(modules -path '*/src/main/*' -name '*.scala' -not -path '*/examples/*') +for file in "${allowlist[@]}"; do + find_args+=(-not -path "*${file#modules/}") +done + +sources=() +while IFS= read -r -d '' source; do + sources+=("$source") +done < <(find "${find_args[@]}" -print0) + +# Hits on `pattern`, with comment lines dropped: the types removed from these APIs are named in the scaladoc +# that explains why they were removed. +hits() { + grep -nE "$1" "${sources[@]}" | grep -vE ':[0-9]+: *(\*|//|/\*)' || true +} + +failed=0 + +report() { + if [ -n "$2" ]; then + echo "$1" >&2 + printf '%s\n' "$2" >&2 + failed=1 + fi +} + +report "\`Any\` in production sources. Give the value a type, or allowlist the file in $0 with a reason:" \ + "$(hits '\bAny\b')" +report "\`asInstanceOf\` in production sources. Make the types line up, or allowlist the file in $0 with a reason:" \ + "$(hits 'asInstanceOf')" + +if [ "$failed" -ne 0 ]; then + exit 1 +fi + +echo "No Any or asInstanceOf in production sources outside the allowlist."